Source Modules

src/graphs

graph_gen

NetworkX-based multi-layer graph generators (legacy, not actively maintained).

This module provides GraphGenerator and several concrete subclasses that build multi-layer contact graphs backed by a NetworkX MultiGraph. These classes pre-date the LightGraph implementation and are no longer used in production simulations; random-graph variants are planned for future use.

Classes:

GraphGenerator: Base class wrapping a NetworkX MultiGraph with layer-aware helper methods. RandomSingleGraphGenerator: Generates a single-layer Barabasi-Albert graph with exponentially trimmed degree distribution. RandomGraphGenerator: Generates one Barabasi-Albert graph per layer with random truncated-normal edge weights. PickleGraphGenerator: Restores a graph from a NetworkX pickle file. CSVGraphGenerator: Builds a graph from nodes, edges and layer CSV files.

class graph_gen.GraphGenerator(random_seed=None)[source]

Bases: object

Base class for multi-layer contact-graph generators.

Wraps a NetworkX MultiGraph (self.G) and provides layer-aware helpers for building, querying and modifying multi-layer contact networks. Subclasses are responsible for populating self.G with nodes and edges.

Class Attributes:
layer_names (list[str]): Ordered list of layer name strings.

Default covers 14 generic layers ('F' through 'Z').

layer_probs (list[float]): Per-layer transmission probabilities,

initialised to 1.0 for every layer.

layer_types (list[int]): Integer indices corresponding to each layer.

G

The underlying multi-layer graph.

Type:

networkx.MultiGraph

Graphs

Optional per-layer single-layer graph cache populated by as_dict_of_graphs().

Type:

dict

A

Cached final adjacency matrix. None until first computed.

Type:

scipy.sparse.csr_matrix or None

A_valid

True when the cached adjacency matrix is up to date with the current edge weights.

Type:

bool

A_invalids

Set of (u, v) node pairs whose adjacency matrix entries need recomputing.

Type:

set

quarantined_edges

Mapping from edge keys to their original weights, used to restore edges after a quarantine is lifted.

Type:

dict

Parameters:

random_seed (int, optional) – Seed for NumPy’s random number generator. None leaves the global RNG state unchanged.

layer_names = ['F', 'D', 'P', 'E', 'H', 'K', 'C', 'S', 'O', 'L', 'R', 'T', 'X', 'Z']
layer_probs = [1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0]
layer_types = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13]
property nodes

NodeView of all nodes in the underlying MultiGraph.

Returns:

All nodes of self.G.

Return type:

networkx.classes.reportviews.NodeView

number_of_nodes()[source]

Return the total number of nodes in the graph.

Returns:

Number of nodes in self.G.

Return type:

int

as_multigraph()[source]

Return the raw underlying NetworkX MultiGraph.

Returns:

The underlying graph object self.G.

Return type:

networkx.MultiGraph

as_one_graph()[source]

Collapse the MultiGraph into a simple undirected Graph.

Parallel edges between any pair of nodes are merged.

Returns:

A simple graph derived from self.G.

Return type:

networkx.Graph

as_aggregated_graph()[source]

Build an aggregated single-layer graph (incomplete, legacy stub).

This method is a historical stub and does not return a fully populated graph. It exists for backward-compatibility only.

Returns:

An empty graph with the same node set as self.G.

Return type:

networkx.Graph

final_adjacency_matrix()[source]

Compute (or return cached) the final contact-probability adjacency matrix.

The matrix entry A[i, j] holds the probability that nodes i and j make contact across any layer, calculated as 1 - product_over_layers(1 - p_layer * w_edge).

A two-level caching strategy is used:

  • If A_valid is True the cached matrix is returned immediately.

  • If the matrix has never been computed (A is None) or A_invalids is empty, the entire matrix is recalculated from scratch.

  • Otherwise only the entries listed in A_invalids are recomputed.

Returns:

Symmetric probability matrix of shape (N, N) where N is the number of nodes.

Return type:

scipy.sparse.csr_matrix

get_graph_for_layer(layer_name)[source]

Extract a single-layer subgraph for the named layer.

Parameters:

layer_name (str) – Name of the layer to extract, must appear in self.layer_names.

Returns:

A new simple graph containing all nodes of self.G and only the edges whose 'type' attribute matches the index of layer_name in self.layer_names. The graph’s metadata includes 'layer_name' and 'layer_prob'.

Return type:

networkx.Graph

as_dict_of_graphs()[source]

Build a per-layer dictionary of single-layer graphs.

Iterates over all layers, constructs a separate networkx.Graph for each one (all nodes, only edges on that layer) and caches the result in self.Graphs.

Returns:

Mapping of layer name to its corresponding single-layer graph.

Return type:

dict[str, networkx.Graph]

get_attr_list(attr)[source]

Return a list of node attribute values in node-iteration order.

Parameters:

attr (str) – Name of the node attribute to collect.

Returns:

Attribute values for every node in self.G, ordered by NetworkX’s default node-iteration order.

Return type:

list

get_edges_for_node(node_id)[source]

Return all edges incident to a given node.

Parameters:

node_id – Node identifier as stored in self.G. If the node does not exist a warning is printed and None is returned implicitly.

Returns:

An edge-data view for all edges incident to node_id, or None (implicit) if the node is not in the graph.

Return type:

networkx.classes.reportviews.OutMultiEdgeDataView or None

get_layers_for_edge(u, v)[source]

Return the list of layer types for all parallel edges between two nodes.

Parameters:

  • u – Source node identifier.

  • v – Destination node identifier.

Returns:

Layer type ('type' attribute) for each parallel edge between u and v in the MultiGraph.

Return type:

list[int]

modify_layers_for_nodes(node_id_list, what_by_what, is_quarrantined=None)[source]

Apply quarantine by scaling edge weights for given nodes.

For each edge incident to a node in node_id_list whose layer appears in what_by_what, the edge weight is multiplied by the corresponding coefficient (clipped to [0, 1]). Edges that are already quarantined (i.e. at least one endpoint has a non-zero quarantine counter in is_quarrantined) are skipped.

The original weight is recorded in self.quarantined_edges so it can be restored by recover_edges_for_nodes(). Affected (u, v) pairs are added to A_invalids so the adjacency matrix will be partially recomputed on next access.

Parameters:

  • node_id_list (iterable) – Node identifiers whose incident edges should be modified.

  • what_by_what (dict) – Mapping of {layer_type: coefficient} specifying the weight multiplier for each layer. If falsy (empty or None) the method returns immediately.

  • is_quarrantined (numpy.ndarray, optional) – Per-node quarantine counters. When provided, edges where either endpoint has a non-zero counter are skipped.

recover_edges_for_nodes(release, normal_life, is_quarrantined)[source]

Restore original edge weights for nodes being released from quarantine.

Iterates over all edges incident to release nodes. For edges where both endpoints have a zero quarantine counter in is_quarrantined, the weight stored in self.quarantined_edges is written back. Edges where at least one endpoint is still quarantined are left unchanged.

Affected (u, v) pairs are added to A_invalids so the adjacency matrix is recomputed on next access.

Parameters:

  • release (iterable) – Node identifiers being released from quarantine.

  • normal_life – Unused parameter retained for interface compatibility with the LightGraph API.

  • is_quarrantined (numpy.ndarray) – Per-node quarantine counters indexed by node ID. Nodes with a counter of 0 are considered free.

get_layers_info()[source]

Return a mapping of layer names to their transmission probabilities.

Returns:

Dictionary mapping each layer name to its current transmission probability as stored in self.G.graph['layer_probs'].

Return type:

dict[str, float]

print_multi()[source]

Print the MultiGraph in DOT format to stdout.

draw_multi(filename='empty_graph.png')[source]

Render the MultiGraph using Graphviz and save it to a file.

Parameters:

filename (str) – Output file path for the rendered image. Defaults to 'empty_graph.png'.

close_layers(list_of_layers, coefs=None)[source]

Set transmission probability to zero (or a custom value) for named layers.

Invalidates the cached adjacency matrix so it will be recomputed on next access.

Parameters:

  • list_of_layers (list[str]) – Names of layers to close.

  • coefs (list[float], optional) – Per-layer replacement probabilities. When provided, coefs[idx] is used for the layer at position idx in list_of_layers. When None all listed layers are set to 0.

write_pickle(path)[source]

Serialize the underlying NetworkX graph to a pickle file.

Parameters:

path (str) – Destination file path for the pickle.

read_picke(path)[source]

Restore the graph from a NetworkX pickle file.

Note: The method name contains a deliberate historical typo (read_picke instead of read_pickle); it is kept as-is for backward compatibility.

Parameters:

path (str) – Path to the pickle file to read.

graph_gen.custom_exponential_graph(base_graph=None, scale=100, min_num_edges=0, m=9, n=None)[source]

Generate a power-law-esque graph by thinning a Barabasi-Albert base graph.

Starts from a Barabasi-Albert preferential-attachment graph (expected to form a single connected component) and then stochastically removes edges from each node so that the resulting degree distribution has an exponential tail rather than a hard minimum of m.

For each node the number of edges to keep is drawn from an exponential distribution with the given scale, clipped to [min_num_edges, current_degree]. Edges not selected for retention are removed.

Parameters:

  • base_graph (networkx.Graph, optional) – Pre-existing graph to thin. When None a fresh Barabasi-Albert graph of n nodes is created.

  • scale (float) – Scale parameter of the exponential distribution used to sample the number of edges to keep per node. Defaults to 100.

  • min_num_edges (int) – Minimum number of edges guaranteed to be kept per node. Defaults to 0.

  • m (int) – Number of edges added per node when constructing the Barabasi-Albert base graph (only relevant when base_graph is None). Defaults to 9.

  • n (int, optional) – Number of nodes for the Barabasi-Albert graph. Required when base_graph is None.

Returns:

The thinned graph.

Return type:

networkx.Graph

Raises:

AssertionError – If base_graph is None and n is also None.

class graph_gen.RandomSingleGraphGenerator(num_nodes=10000, **kwargs)[source]

Bases: GraphGenerator

Graph generator producing a single-layer random contact graph.

Builds one Barabasi-Albert graph (with exponential degree thinning) and assigns each edge a truncated-normal weight in (0, 1). The resulting graph has no layer structure (it collapses all contacts into a single layer).

Parameters:

  • num_nodes (int) – Number of nodes. Defaults to 10000.

  • **kwargs – Additional keyword arguments forwarded to GraphGenerator.

class graph_gen.RandomGraphGenerator(num_nodes=10000, **kwargs)[source]

Bases: GraphGenerator

Graph generator producing independent random contact graphs per layer.

For each of the 14 default layers a separate Barabasi-Albert graph is generated (with exponential degree thinning) and each edge is assigned a truncated-normal weight in (0, 1) with mean 0.7 and std-dev 0.3. Layer-level transmission probabilities are also drawn from the same truncated-normal distribution. All per-layer graphs are merged into the shared MultiGraph self.G.

The resulting graph has a mean degree of approximately 13 per layer.

Parameters:

  • num_nodes (int) – Number of nodes in each per-layer graph. Defaults to 10000.

  • **kwargs – Additional keyword arguments forwarded to GraphGenerator.

as_dict_of_graphs()[source]

Return the pre-built per-layer graph dictionary.

Overrides GraphGenerator.as_dict_of_graphs() to return the dictionary built during __init__ rather than rebuilding it.

Returns:

Mapping of layer name to its single-layer graph.

Return type:

dict[str, networkx.Graph]

class graph_gen.PickleGraphGenerator(path_to_pickle='graph.pickle', **kwardgs)[source]

Bases: GraphGenerator

Graph generator that restores a graph from a NetworkX pickle file.

Parameters:

  • path_to_pickle (str) – Path to the pickle file containing a previously serialized NetworkX graph. Defaults to 'graph.pickle'.

  • **kwardgs – Additional keyword arguments (currently unused).

class graph_gen.CSVGraphGenerator(path_to_nodes='nodes.csv', path_to_edges='edges.csv', path_to_layers='etypes.csv', **kwargs)[source]

Bases: GraphGenerator

Graph generator that builds a multi-layer graph from CSV files.

Reads nodes, edges and layer definitions from three separate CSV files and populates the underlying NetworkX MultiGraph. Duplicate edges (same type, subtype, endpoint pair and weight) are removed in debug mode.

Class Attributes:

layer_names (list): Starts empty; populated from the layers CSV. layer_probs (list): Starts empty; populated from the layers CSV.

Parameters:

  • path_to_nodes (str) – Path to the nodes CSV file. Defaults to 'nodes.csv'.

  • path_to_edges (str) – Path to the edges CSV file. Defaults to 'edges.csv'. Required columns: vertex1, vertex2, type, subtype, weight.

  • path_to_layers (str) – Path to the layer-definition CSV file. Defaults to 'etypes.csv'. Required columns: id, name, weight.

  • **kwargs – Additional keyword arguments forwarded to GraphGenerator.

layer_names = []
layer_probs = []
__str__()[source]

Return a human-readable string listing all edges with their data.

Returns:

Newline-separated representation of every edge in self.G including edge attribute dictionaries.

Return type:

str

light

Light-weight graph representation for agent-based network simulations.

This module provides the LightGraph class, which stores a multi-layer undirected graph as a collection of NumPy arrays and a compressed sparse row (CSR) adjacency matrix. It is designed to be considerably faster than NetworkX for the high-throughput edge queries required by epidemic and information-spread models.

Typical workflow:

  1. Build a LightGraph from CSV files with LightGraph.read_csv().

  2. Pickle the result for repeated simulation runs.

  3. Clone per-run state cheaply with LightGraph.copy().

light.concat_lists(l)[source]

Return the concatenation of all lists contained in an iterable.

Parameters:

l (iterable) – An iterable whose elements are lists (or other iterables) to be concatenated.

Returns:

A single flat list produced by chaining every element of l.

Return type:

list

class light.LightGraph(random_seed=None)[source]

Bases: object

Graph for agent-based network models stored as NumPy arrays.

NetworkX proved too slow for the high-throughput edge queries needed by large-scale simulations. LightGraph therefore stores the graph as a collection of plain NumPy arrays together with a CSR adjacency matrix whose non-zero values are indices into those arrays.

Typical usage is to load the graph from CSV files once with read_csv() and then pickle the result for future simulation runs, because building from CSV is comparatively slow.

Nodes carry integer IDs that are user-visible, but internally (and in the model) nodes are addressed by their positional index in the original nodes file.

random_seed

The seed passed to numpy.random.seed at construction time, stored for reproducibility bookkeeping.

Type:

int or None

edge_repo

Internal edge repository (populated by read_csv()).

A

Adjacency matrix whose non-zero entries hold edge-repository keys. None until read_csv() has been called.

Type:

scipy.sparse.csr_matrix or None

is_quarantined

Per-node quarantine counter. None until the first quarantine operation.

Type:

numpy.ndarray or None

Parameters:

random_seed (int, optional) – Seed for NumPy’s random number generator. None leaves the global state unchanged.

read_csv(path_to_nodes='p.csv', path_to_external=None, path_to_layers=None, path_to_edges='edges.csv', path_to_quarantine=None, path_to_layer_groups=None)[source]

Load graph data from CSV files and build internal data structures.

Reads node attributes, edge lists, layer definitions, optional external nodes, optional quarantine coefficients and optional layer groups. After this call the graph is ready to use.

Self-loops are silently dropped with a warning.

Parameters:

  • path_to_nodes (str) – Path to the primary nodes CSV file. Defaults to 'p.csv'.

  • path_to_external (str, optional) – Path to an optional secondary nodes CSV file (e.g. external / virtual nodes). When None no external nodes are added.

  • path_to_layers (str, optional) – Path to a CSV file defining layer IDs, names and weights. When None a minimal two-layer default is used (layers 0 and 1).

  • path_to_edges (str) – Path to the edges CSV file. Defaults to 'edges.csv'. Required columns are vertex1, vertex2; optional columns are layer, sublayer, probability and intensity (all back-filled with sensible defaults when absent).

  • path_to_quarantine (str, optional) – Path to a two-column headerless CSV file mapping layer ID to quarantine coefficient. When None quarantine coefs are disabled.

  • path_to_layer_groups (str, optional) – Path to a JSON file defining named groups of layers. When None layer groups are disabled.

property number_of_nodes

Total number of nodes in the graph (base + external).

Returns:

Number of nodes.

Return type:

int

get_nodes(layer)[source]

Return all node indices that have at least one edge on the given layer.

Parameters:

layer (int) – Layer ID to filter by.

Returns:

Sorted array of unique node indices that participate in at least one edge whose layer equals layer.

Return type:

numpy.ndarray

get_edges_nodes(edges, edges_dirs)[source]

Return the source and destination node indices for a set of edges.

Edges are stored with an arbitrary orientation (smaller index as source). edges_dirs corrects for this: when True the stored source is the logical source; when False the stored destination is the logical source.

Parameters:

  • edges (numpy.ndarray) – 1-D array of edge indices.

  • edges_dirs (numpy.ndarray) – Boolean array of the same length as edges. True means forward direction (source → dest); False means backward direction.

Returns:

A pair (source_nodes, dest_nodes) where each element is a 1-D integer array of node indices aligned with edges.

Return type:

tuple[numpy.ndarray, numpy.ndarray]

get_edges(source_flags, dest_flags, dirs=True)[source]

Return all edges between two sets of nodes.

Parameters:

  • source_flags (numpy.ndarray) – Binary vector of length num_nodes. A value of 1 marks nodes in the first (source) set.

  • dest_flags (numpy.ndarray) – Binary vector of length num_nodes. A value of 1 marks nodes in the second (destination) set.

  • dirs (bool) – When True (default) also return direction flags alongside the edge indices.

Returns:

If dirs is True, returns (edges, directions) where edges is a 1-D integer array of edge indices and directions is a corresponding boolean array (True = forward). If dirs is False, only edges is returned. Both arrays are empty when no edges exist between the two sets.

Return type:

tuple[numpy.ndarray, numpy.ndarray] or numpy.ndarray

get_nodes_edges(nodes)[source]

Return all edge indices adjacent to a set of nodes.

Parameters:

nodes (array-like) – Node indices whose incident edges are requested.

Returns:

Flat list of edge indices (possibly with duplicates when multiple nodes share an edge). Returns an empty list when nodes is empty or when none of the nodes have any edges.

Return type:

list

get_nodes_edges_on_layers(nodes, layers)[source]

Return incident edge indices for a set of nodes filtered by layer.

Identical to get_nodes_edges() but restricts the result to edges whose layer ID appears in layers.

Parameters:

  • nodes (array-like) – Node indices whose incident edges are requested.

  • layers (array-like) – Allowed layer IDs. Only edges on one of these layers are included in the result.

Returns:

Flat list of edge indices that belong to one of the specified layers. Returns an empty list when no matching edges are found.

Return type:

list

switch_off_edges(edges)[source]

Permanently deactivate a list of edges.

Sets e_active to False for each edge in edges. This operation is independent of quarantine state and sets the effective transmission probability to zero regardless of other factors.

Parameters:

edges (list) – List of edge indices to deactivate.

switch_on_edges(edges)[source]

Reactivate a list of previously deactivated edges.

Reverses switch_off_edges() by setting e_active back to True. Does not interact with quarantine state.

Parameters:

edges (list) – List of edge indices to reactivate.

get_all_edges_probs()[source]

Compute effective transmission probabilities for every edge.

Combines the raw per-edge probability (e_probs), the quarantine validity flag (e_valid), the active flag (e_active) and the layer weight into a single scalar per edge. Uses numexpr for fast vectorised evaluation.

When e_valid == 2 the edge is fully active and the probability is e_probs * layer_weight. Any other value of e_valid is interpreted as an override probability (e.g. from quarantine) and replaces e_probs directly.

Returns:

Float array of length n_edges with effective transmission probabilities in [0, 1].

Return type:

numpy.ndarray

get_edges_probs(edges)[source]

Compute effective transmission probabilities for a subset of edges.

Applies the same logic as get_all_edges_probs() but only for the requested edge indices.

Parameters:

edges (numpy.ndarray) – 1-D integer array of edge indices.

Returns:

Float array of the same length as edges with effective transmission probabilities in [0, 1].

Return type:

numpy.ndarray

get_edges_intensities(edges)[source]

Return the intensity values for a subset of edges.

Parameters:

edges (numpy.ndarray) – 1-D integer array of edge indices.

Returns:

Float array of the same length as edges containing the raw intensity value (e_intensities) for each requested edge.

Return type:

numpy.ndarray

is_super_edge(edges)[source]

Identify super-spreader edges (Hodonin layer convention).

Super-spreader edges are defined as those with layer ID >= 33.

Parameters:

edges (numpy.ndarray) – 1-D integer array of edge indices.

Returns:

Boolean array of the same length as edges, True where the edge is a super-spreader edge.

Return type:

numpy.ndarray

is_family_edge(edges)[source]

Identify household/family edges (Hodonin layer convention).

Family edges are those with layer ID 1 or 2.

Parameters:

edges (numpy.ndarray) – 1-D integer array of edge indices.

Returns:

Boolean array of the same length as edges, True where the edge belongs to a family layer.

Return type:

numpy.ndarray

is_class_edge(edges, all=True)[source]

Identify school/class edges (Hodonin layer convention).

School edges have layer IDs 4–11. The all flag controls whether all school types are included or only lower school levels (layers 4–7, excluding high school and higher elementary).

Parameters:

  • edges (numpy.ndarray) – 1-D integer array of edge indices.

  • all (bool) – When True (default), include all school layers (4–11). When False, restrict to layers 4–7.

Returns:

Boolean array of the same length as edges, True where the edge is a class/school edge.

Return type:

numpy.ndarray

is_pub_edge(edges)[source]

Identify pub/bar edges (Hodonin layer convention).

Pub edges have layer IDs 20, 28 or 29.

Parameters:

edges (numpy.ndarray) – 1-D integer array of edge indices.

Returns:

Boolean array of the same length as edges, True where the edge represents a pub/bar contact.

Return type:

numpy.ndarray

modify_layers_for_nodes(node_id_list, what_by_what)[source]

Apply quarantine by reducing edge probabilities for given nodes.

For every edge incident to a node in node_id_list that is not already quarantined (e_valid == 2), the effective probability is replaced by e_probs * coef (clipped to [0, 1]), where coef is the coefficient for that edge’s layer taken from what_by_what.

The per-node quarantine counter is_quarantined is incremented so that edges shared with still-quarantined nodes are not accidentally restored by a partial release.

Parameters:

  • node_id_list (array-like) – Positional indices of nodes to quarantine.

  • what_by_what (dict) – Mapping of {layer_id: coefficient} specifying the multiplicative reduction per layer. Must not be empty.

Raises:

ValueError – If what_by_what is falsy (empty or None).

recover_edges_for_nodes(release)[source]

Restore original edge probabilities when nodes leave quarantine.

Decrements the quarantine counter for each node in release. For nodes whose counter reaches zero (i.e. no longer quarantined on any count), edges that connect two fully released nodes have their e_valid flag reset to 2 (fully active). Edges that still touch a quarantined node are left unchanged.

Parameters:

release (numpy.ndarray) – Positional indices of nodes being released from quarantine.

final_adjacency_matrix()[source]

Return self for backward compatibility with older graph interfaces.

Earlier versions of the model expected a separate adjacency-matrix object. This shim allows code written against that interface to work without modification.

Returns:

The graph instance itself.

Return type:

LightGraph

get_layer_for_edge(e)[source]

Return the layer ID for a single edge.

Parameters:

e (int) – Edge index.

Returns:

Layer ID of edge e.

Return type:

int

set_layer_weights(weights)[source]

Replace the layer-weight vector and rebuild per-edge weight cache.

The per-edge weight array e_layer_weight is recomputed in-place after the update so that subsequent probability calculations use the new weights immediately.

Parameters:

weights (array-like) – New weight values indexed by layer ID. Must be the same length as the current layer_weights array.

close_layers(list_of_layers, coefs=None)[source]

Set the weight of named layers to zero (or a custom coefficient).

Useful for simulating non-pharmaceutical interventions such as school or workplace closures.

Parameters:

  • list_of_layers (list[str]) – Names of layers to close, matched against self.layer_name.

  • coefs (list[float], optional) – Per-layer replacement weights. When provided, coefs[idx] is used instead of 0 for the layer at position idx in list_of_layers. When None all listed layers are set to weight 0.

copy()[source]

Return an optimised partial copy of the graph for a new simulation run.

The vast majority of graph fields (topology, node attributes, edge metadata) are immutable between runs and are therefore shared via a shallow copy. Only the mutable, run-specific arrays—e_valid, layer_weights, e_active and e_layer_weight—are deep- copied and reset to their default values. The quarantine tracker is_quarantined is reset to None.

Returns:

A new LightGraph instance that shares topology with the original but has independent mutable state.

Return type:

LightGraph

romeo_juliet_graph_gen

Romeo and Juliet test graph for small-scale simulation experiments.

This module defines RomeoAndJuliet, a handcrafted multi-layer contact graph based on the characters and social relationships in Shakespeare’s Romeo and Juliet. It is intended purely as a small, interpretable test fixture for validating model mechanics and is not used in production simulations.

The graph uses 15 layer types (indices 0–14) matching the GraphGenerator convention, where meaningful layers are family (1), friends (10), work (11), work to clients (12), public transport (13) and shops and events (14).

class romeo_juliet_graph_gen.RomeoAndJuliet(**kwargs)[source]

Bases: GraphGenerator

Handcrafted multi-layer contact graph of the Romeo and Juliet cast.

Encodes the social network of all named characters (35 nodes) across five meaningful contact layers:

  • Layer 1 – family

  • Layer 10 – friends

  • Layer 11 – work

  • Layer 12 – work to clients

  • Layer 13 – public transport

  • Layer 14 – shops and events (Capulet party subgraph)

Edge weights are assigned randomly from a truncated-normal distribution TruncNorm(mu=0.7, sigma=0.3, lower=0, upper=1) for each edge.

This class is intended as a small, interpretable test fixture and is not used in production simulations.

Class Attributes:
layer_probs (list[int]): Transmission probability per layer,

all initialised to 1.

layer_names (list[str]): Human-readable name for each of the 15

layer indices.

Parameters:

**kwargs – Keyword arguments forwarded to GraphGenerator.

layer_probs = [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1]
layer_names = ['not in Verona', 'family', 'not in Verona', 'not in Verona', 'not in Verona', 'not in Verona', 'not in Verona', 'not in Verona', 'not in Verona', 'not in Verona', 'friends', 'work', 'work to clients', 'public transport', 'shops and events']
write_to_csv(prefix='raj')[source]

Export the graph to three CSV files (edges, nodes, layer types).

The three output files are named:

  • <prefix>-edges.csv – columns: vertex1, vertex2, type, weight

  • <prefix>-nodes.csv – columns: type, label, sex, age

  • <prefix>-etypes.csv – columns: id, name, weight

Parameters:

prefix (str) – Common filename prefix for the output files. Defaults to 'raj'.

simple

Minimal directed graph for information-spread models.

This module provides SimpleGraph, a lightweight directed graph representation designed for information models (e.g. opinion dynamics, rumour spreading) where edges are oriented and all transmission probabilities are uniform. The graph is stored as two NumPy arrays—one for edge sources and one for edge destinations—keeping memory overhead minimal.

class simple.SimpleGraph(random_seed=None)[source]

Bases: object

Lightweight directed graph for information models.

The graph stores only directed edges (no layer information, no per-edge probabilities) as two NumPy arrays of source and destination node indices. This is sufficient for simple information/opinion models where all contacts are equally likely to transmit.

random_seed

Seed supplied at construction, stored for reproducibility bookkeeping.

Type:

int or None

e_source

Source node index for each edge.

Type:

numpy.ndarray

e_dest

Destination node index for each edge.

Type:

numpy.ndarray

num_nodes

Total number of nodes inferred from the maximum node index seen in the edge list.

Type:

int

Parameters:

random_seed (int, optional) – Seed for NumPy’s random number generator. None leaves the global RNG state unchanged.

read_csv(path_to_edges='edges.csv')[source]

Load a directed edge list from a CSV (or whitespace-separated) file.

The file must contain exactly two columns—source node and destination node—either comma- or whitespace-separated, with no header row. Self-loops are dropped with a warning logged at the WARNING level. The node count is inferred as max(source, dest) + 1.

Parameters:

path_to_edges (str) – Path to the edge-list file. Defaults to 'edges.csv'.

property number_of_nodes

Total number of nodes in the graph.

Returns:

Number of nodes (inferred from the maximum node index in the edge list, plus one).

Return type:

int

copy()[source]

Return a shallow copy of the graph.

Because SimpleGraph has no mutable run-time state (unlike LightGraph) a plain shallow copy is sufficient to produce an independent graph object for a new simulation run.

Returns:

A shallow copy of self.

Return type:

SimpleGraph

src/model_m

model_m

src/models

agent_based_network_model

Agent-based epidemic network model (SimulationDrivenModel).

This module defines SimulationDrivenModel, the primary agent-based epidemiological model of the MAIS project. It inherits from SimulationEngine and implements:

  • Stochastic duration sampling for every disease stage using pre-loaded probability distributions (JSON file).

  • Age- and sex-stratified mortality via a CSV death-probability table.

  • Network-contact infection via prob_of_contact().

  • Support for external (EXT) nodes with controllable edge activity.

  • Utility methods for scenario manipulation (move_to_E, move_to_R, etc.) and output statistics (df_source_infection, df_source_nodes).

class agent_based_network_model.SimulationDrivenModel(G, **kwargs)[source]

Bases: SimulationEngine

Agent-based SEIR-variant epidemic model with stochastic duration sampling.

Each agent (network node) progresses through disease compartments with durations drawn from empirical distributions. Key features:

  • Compartments: S, S_s, E, I_n, I_a, I_s, J_s, J_n, R, D, EXT.

  • Duration distributions loaded from a JSON file at initialisation.

  • Age/sex-stratified case-fatality rates from a CSV file.

  • Daily infection driven by prob_of_contact().

  • Optional external-node mechanism (EXT state) with probabilistic edge activation.

Class-level attributes define the model structure (states, transitions, parameters). All are overridable by subclasses.

states = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
num_states = 11
state_str_dict = {0: 'S', 1: 'S_s', 2: 'E', 3: 'I_n', 4: 'I_a', 5: 'I_s', 6: 'J_s', 7: 'J_n', 8: 'R', 9: 'D', 10: 'EXT'}
ext_code = 10
transitions = [(1, 2), (1, 0), (0, 2), (0, 1), (2, 3), (2, 4), (3, 7), (4, 5), (5, 6), (6, 8), (6, 9), (7, 8)]
num_transitions = 12
final_states = [8, 9]
invisible_states = [9, 10]
unstable_states = [2, 3, 4, 5, 7, 6]
fixed_model_parameters = {'beta_reduction': (0, 'reduction of beta for asymptomatic multiplier'), 'durations_file': ('../config/duration_probs.json', 'file with probs for durations'), 'ext_epi': (0, 'prob of beeing infectious for external nodes'), 'false_symptoms_rate': (0, ''), 'false_symptoms_recovery_rate': (1.0, ''), 'mu': (0, 'rate of infection-related death'), 'p': (0, 'probability of interaction outside adjacent nodes'), 'prob_death_file': ('../data/prob_death.csv', 'file with probs for durations'), 'q': (0, ' probability of detected individuals interaction outside adjacent nodes'), 'save_nodes': (False, '')}
model_parameters = {'asymptomatic_rate': (0, 'asymptomatic rate'), 'beta': (0, 'rate of transmission (exposure)'), 'beta_A': (0, 'hidden parameter'), 'beta_A_in_family': (0, 'hidden parameter'), 'beta_in_family': (0, 'hidden parameter'), 'infectious_time': (-1, 'time_from first_symptom  - do not setup'), 'psi_E': (0, 'probability of positive test results for exposed individuals'), 'psi_Ia': (0, 'probability of positive test results for Ia individuals'), 'psi_In': (0, 'probability of positive test results for In individuals'), 'psi_Is': (0, 'probability of positive test results for Is individuals'), 'symptomatic_time': (-1, 'time_from first_symptom  - do not setup'), 'test_rate': (1.0, 'test rate'), 'theta_E': (0, 'rate of baseline testing for exposed individuals'), 'theta_Ia': (0, 'rate of baseline testing for Ia individuals'), 'theta_In': (0, 'rate of baseline testing for In individuals'), 'theta_Is': (0, 'rate of baseline testing for Is individuals')}
inicialization()[source]

Initialise the model: set derived beta parameters, load duration and death data.

Before calling the parent initialiser, sets beta_in_family, beta_A, and beta_A_in_family in init_kwargs based on the supplied beta and beta_reduction.

After parent init:

  • Allocates self.testable and self.will_die arrays.

  • Loads duration-probability distributions from self.durations_file (JSON).

  • Loads age/sex death probabilities from self.prob_death_file (CSV).

setup_series_and_time_keeping()[source]

Extend time-keeping setup with infection-time and contact-history buffers.

Adds:

  • self.infect_time – per-node infectious-period counter.

  • self.contact_history – 14-day rolling contact buffer.

  • self.successfull_source_of_infection – per-node count of successful transmissions originated.

  • self.stat_successfull_layers – per-layer daily transmission counts.

states_and_counts_init(ext_nodes=None, ext_code=None)[source]

Extend state initialisation with per-node disease-timeline arrays.

After calling the parent states_and_counts_init():

  • Allocates infectious_time, symptomatic_time, rna_time, and time_to_die arrays (all initialised to -1).

  • Resets self.testable.

  • Sets self.need_check for S / S_s nodes.

  • Calls update_plan() for all nodes to set initial plans.

Parameters:

  • ext_nodes (int, optional) – Number of external nodes.

  • ext_code (int, optional) – State code for external nodes.

daily_update(nodes)[source]

Perform daily infection checks and update plans for susceptible nodes.

For susceptible (S / S_s) nodes:

  1. Optionally activates external-node edges (if external nodes present).

  2. Calls prob_of_contact() to compute per-node exposure probabilities.

  3. Sets time_to_go=1 and state_to_go=E for newly exposed nodes.

Parameters:

nodes (numpy.ndarray) – Boolean bitmap of nodes that need a daily check (from self.need_check).

update_plan(nodes)[source]

Generate new transition plans for nodes that just changed state.

For each state, samples the appropriate duration(s) from the loaded duration distributions and sets self.time_to_go, self.state_to_go, and self.need_check accordingly.

State-specific logic:

  • S: no scheduled transition; flagged for daily checks.

  • E: samples incubation duration; branches to I_n or I_a stochastically based on asymptomatic_rate.

  • I_n: samples infectious + RNA-positivity durations; schedules J_n.

  • I_a: samples asymptomatic, infectious, and RNA durations; schedules I_s.

  • I_s: decides death outcome; schedules J_s or D.

  • J_s / J_n: schedules R (or D for dying nodes).

  • R / D: clears plan.

Parameters:

nodes (numpy.ndarray) – Boolean bitmap of nodes whose plans should be regenerated.

run_iteration()[source]

Perform one day of simulation with infection-time tracking.

Before delegating to the parent run_iteration():

  • Resets per-layer successful-transmission counters for the current day.

  • Increments self.infect_time for currently infectious nodes.

  • When in debug mode, verifies that external-node state constraints are satisfied.

move_to_R(nodes)[source]

Move a list of nodes to the Recovered state.

Parameters:

nodes (array-like of int) – Node indices to recover.

move_target_nodes_to_R(target_nodes)[source]

Move nodes (given as a boolean bitmap) to the Recovered state.

Parameters:

target_nodes (numpy.ndarray) – Boolean bitmap of shape (num_nodes,); nodes where True are moved to R.

move_target_nodes_to_S(target_nodes)[source]

Move nodes (given as a boolean bitmap) to the Susceptible state.

Parameters:

target_nodes (numpy.ndarray) – Boolean bitmap of shape (num_nodes,); nodes where True are moved to S.

move_to_E(num)[source]

Randomly expose num susceptible (S or S_s) nodes by moving them to E.

Selects nodes uniformly at random from all currently susceptible nodes. If fewer susceptible nodes are available than requested, logs a warning and returns without action.

Parameters:

num (int) – Number of nodes to expose.

df_source_infection()[source]

Return a DataFrame of successful transmissions per contact layer per day.

Returns:

Indexed by day (0..t-1), with one column per graph layer named after the layer.

Return type:

pandas.DataFrame

df_source_nodes()[source]

Return a Series of successful infection counts per infectious node.

Filters to nodes that are not in S, S_s, or E (i.e. nodes that were at some point infectious).

Returns:

Successful infection counts indexed by node.

Return type:

pandas.Series

die_or_not_to_die(target_nodes)[source]

Decide stochastically whether each target node will die from the disease.

Uses age- and sex-stratified case-fatality rates scaled by self.mu.

Parameters:

target_nodes (numpy.ndarray) – Boolean bitmap of nodes entering the symptomatic state.

Returns:

Boolean array of length target_nodes.sum(); True where the node is destined to die.

Return type:

numpy.ndarray

get_time_to_die(target_nodes)[source]

Sample the number of days until death for each dying node.

Uses a mixed piecewise distribution:

  • If U < 0.571: X = ceil(10 * U / 0.571)

  • Otherwise: X = round(4 - ln(1 - U) / 0.13)

Parameters:

target_nodes (numpy.ndarray) – Boolean bitmap of nodes that have been determined to die (from die_or_not_to_die()).

Returns:

Integer array of length target_nodes.sum() with days-until-death for each dying node.

Return type:

numpy.ndarray

get_dead()[source]

Return summary statistics about deceased nodes.

Returns:

(total_dead, young, old1, old2) where:

  • total_dead (int) – total count of deceased nodes.

  • young (int) – deceased under 65.

  • old1 (int) – deceased aged 65–79.

  • old2 (int) – deceased aged 80+.

Return type:

tuple

flip_coin_for_external_edges()[source]

Stochastically activate external-node edges for the current day.

Switches all external-node edges on (resetting from the previous day), then switches off each edge independently with probability 1 - self.ext_epi. This controls the likelihood that external (imported) cases interact with the main population on any given day.

agent_info_models

Information-diffusion and rumour-spreading model classes.

This module defines several simulation models that reuse the SimulationEngine infrastructure for information / opinion spreading rather than epidemic disease:

class agent_info_models.STATES[source]

Bases: object

Simple SIR state codes for information-diffusion models.

S

Susceptible (uninformed).

Type:

int

I

Infected / informed (spreading the rumour).

Type:

int

R

Recovered (stopped spreading).

Type:

int

EXT

External node.

Type:

int

S = 0
I = 1
R = 2
EXT = 10
class agent_info_models.Tipping[source]

Bases: object

State codes for the threshold-based tipping-point model.

S

Susceptible (not yet adopted).

Type:

int

ACTIVE

Active (adopted / tipped).

Type:

int

EXT

External node.

Type:

int

S = 0
ACTIVE = 1
EXT = 10
class agent_info_models.RumourModel(G, **kwargs)[source]

Bases: SimulationEngine

Basic SIR rumour-spreading model on a contact network.

Agents transition S → I when a neighbour is I (with rate lambda0), and I → R after a fixed duration I_duration.

Inherits the plan-based stepping logic from SimulationEngine.

states = [0, 1, 2, 10]
num_states = 4
state_str_dict = {0: 'S', 1: 'I', 2: 'R', 10: 'EXT'}
ext_code = 10
transitions = [(0, 1), (1, 2)]
num_transitions = 2
final_states = [2]
invisible_states = [10]
unstable_states = [1]
fixed_model_parameters = {'I_duration': (1, 'time in the I state'), 'beta': (0, 'rate of transmission (exposure)')}
inicialization()[source]

Delegate to parent initialiser (no extra setup needed).

setup_series_and_time_keeping()[source]

Delegate to parent time-series setup.

states_and_counts_init(ext_nodes=None, ext_code=None)[source]

Initialise counts and flag S nodes for daily checks.

Parameters:

  • ext_nodes (int, optional) – Number of external nodes.

  • ext_code (int, optional) – State code for external nodes.

prob_of_contact(source_state, dest_state, beta)[source]

Return nodes exposed by direct contact with I-state neighbours.

Iterates over all undirected edges. For each edge between a node in source_state and a node in dest_state, draws a Bernoulli coin with probability beta. Returns the indices of source nodes that were exposed (may contain duplicates if a node has multiple I-neighbours).

Parameters:

  • source_state (int) – State code of susceptible nodes.

  • dest_state (int) – State code of infectious nodes.

  • beta (float) – Per-edge transmission probability.

Returns:

1-D array of exposed node indices (possibly with duplicates).

Return type:

numpy.ndarray

daily_update(nodes)[source]

Perform daily rumour-spreading check for susceptible nodes.

Calls prob_of_contact() and schedules exposed S nodes to move to I on the next day.

Parameters:

nodes (numpy.ndarray) – Boolean bitmap of nodes needing a check.

update_plan(nodes)[source]

Set transition plans for nodes that just changed state.

  • S: no scheduled transition; flagged for daily checks.

  • I: scheduled to move to R after I_duration days.

  • R: no further transition.

Parameters:

nodes (numpy.ndarray) – Boolean bitmap of nodes to update.

run_iteration()[source]

Delegate to parent run_iteration (no extra logic needed).

class agent_info_models.RumourModelInfo(G, **kwargs)[source]

Bases: RumourModel

Extended rumour model with time-decaying transmission and event boost.

Extends RumourModel with:

  • A time-decaying per-node transmission rate: lambda = lambda0 * exp(-scale * time_in_I).

  • An optional event at time t_event that adds a temporary boost to lambda: lambda += event_boost * exp(-decay * (t - t_event)).

  • A stochastic I → R transition each day (probability beta_duration).

The I state is maintained indefinitely until a daily Bernoulli trial (probability beta_duration) triggers recovery.

fixed_model_parameters = {'beta_duration': (0.02, 'probability of ending the I state at each step'), 'decay': (0.06, 'decay rate of the increased spread after event'), 'event_boost': (0.02, 'boost in transmission rate after event'), 'init_I': (12, 'initial number of infected nodes'), 'lambda0': (0.001, 'base rate of transmission'), 'scale': (1.0, 'scaling factor for the transmission probability'), 't_event': (95, 'time of the event that increases the spread')}
prob_of_contact(source_state, dest_state, beta)[source]

Return newly exposed nodes using time-decaying lambda and event boost.

Computes a per-node effective transmission rate lambda_ based on:

  • How long each I node has been infectious (decay over time).

  • Whether the simulation has passed t_event (adds a boost).

Then for each S node uses the compound probability formula 1 - (1 - lambda_)^k where k is its count of I neighbours.

Parameters:

  • source_state (int) – State code of susceptible nodes (S).

  • dest_state (int) – State code of infectious nodes (I).

  • beta – Unused (included for API compatibility with parent).

Returns:

1-D array of exposed node indices.

Return type:

numpy.ndarray

daily_update(nodes)[source]

Perform daily spreading check and stochastic I → R transition.

Calls the parent RumourModel.daily_update(), then for each I node draws a Bernoulli coin (probability beta_duration) to decide if it recovers today.

Parameters:

nodes (numpy.ndarray) – Boolean bitmap of nodes needing a check.

update_plan(nodes)[source]

Set transition plans for RumourModelInfo nodes.

  • S: no scheduled transition; flagged for daily checks.

  • I: no fixed deadline; flagged for daily checks (recovery is decided stochastically each day).

  • R: no further transition.

Parameters:

nodes (numpy.ndarray) – Boolean bitmap of nodes to update.

class agent_info_models.InfoSIRModel(G, **kwargs)[source]

Bases: SimulationEngine

SIR information-spreading model driven by graph edge probabilities.

Uses the full edge-probability machinery from the graph object (rather than the simple per-edge coin flip of RumourModel) to activate contacts each day. The effective transmission rate per active edge is given by the per-node beta parameter multiplied by the edge intensity.

Compartments: S (uninformed), I (spreading), R (stopped), EXT (external).

states = [0, 1, 2, 10]
num_states = 4
state_str_dict = {0: 'S', 1: 'I', 2: 'R', 10: 'EXT'}
ext_code = 10
transitions = [(0, 1), (1, 2)]
num_transitions = 2
final_states = [2]
invisible_states = [10]
unstable_states = [1]
fixed_model_parameters = {'I_duration': (1, 'time in the I state')}
model_parameters = {'beta': (0, 'rate of transmission (exposure)')}
inicialization()[source]

Delegate to parent initialiser (no extra setup needed).

setup_series_and_time_keeping()[source]

Delegate to parent time-series setup.

states_and_counts_init(ext_nodes=None, ext_code=None)[source]

Initialise counts and flag S nodes for daily infection checks.

Parameters:

  • ext_nodes (int, optional) – Number of external nodes.

  • ext_code (int, optional) – State code for external nodes.

prob_of_contact(source_state, dest_state, beta)[source]

Compute per-node exposure probability using edge probabilities and intensities.

Activates edges stochastically using the graph’s edge-probability vector, then evaluates both directions for each active edge. For edges connecting a source_state node to a dest_state node, applies a Bernoulli trial with rate beta[source] * intensity.

Parameters:

  • source_state (int) – State code of susceptible nodes.

  • dest_state (int) – State code of infectious nodes.

  • beta (numpy.ndarray) – Per-node transmission rate array.

Returns:

Binary exposure vector of shape (num_nodes, 1); 1 where a node was newly exposed.

Return type:

numpy.ndarray

daily_update(nodes)[source]

Perform daily infection check for susceptible nodes.

Calls prob_of_contact() and schedules exposed S nodes to move to I on the next day. External nodes are not yet supported.

Parameters:

nodes (numpy.ndarray) – Boolean bitmap of nodes needing a check.

Raises:

NotImplementedError – If external nodes are present.

update_plan(nodes)[source]

Set transition plans for InfoSIRModel nodes.

  • S: no scheduled transition; flagged for daily checks.

  • I: scheduled to move to R after I_duration days.

  • R: no further transition.

Parameters:

nodes (numpy.ndarray) – Boolean bitmap of nodes to update.

run_iteration()[source]

Delegate to parent run_iteration (no extra logic needed).

class agent_info_models.InfoTippingModel(G, **kwargs)[source]

Bases: SimulationEngine

Threshold-based (tipping-point) adoption model on a contact network.

An agent adopts (S → ACTIVE) when the weighted fraction of its active neighbours exceeds its personal threshold theta. Once active, the agent remains active indefinitely.

Edge weights are the graph’s e_intensities values, and daily contact is stochastic (activated by edge probabilities).

states = [0, 1, 10]
num_states = 3
state_str_dict = {0: 'S', 1: 'Active', 10: 'Ext'}
ext_code = 10
transitions = [(0, 1)]
num_transitions = 1
model_parameters = {'theta': (0, 'threshold')}
states_and_counts_init(ext_nodes=None, ext_code=None)[source]

Initialise counts and flag S nodes for daily tipping checks.

Parameters:

  • ext_nodes (int, optional) – Number of external nodes.

  • ext_code (int, optional) – State code for external nodes.

daily_update(nodes)[source]

Perform daily tipping check: activate nodes whose threshold is met.

Calls _transmission() and schedules newly tipping nodes to move to ACTIVE on the next day. External nodes are not supported.

Parameters:

nodes (numpy.ndarray) – Boolean bitmap of nodes needing a check.

Raises:

NotImplementedError – If external nodes are present.

update_plan(nodes)[source]

Set transition plans for InfoTippingModel nodes.

  • S: no scheduled transition; flagged for daily tipping checks.

  • ACTIVE: stays ACTIVE indefinitely; no further scheduled transition.

Parameters:

nodes (numpy.ndarray) – Boolean bitmap of nodes to update.

engine

Base simulation engine for epidemic network models.

This module defines BaseEngine, which provides the common infrastructure shared by all concrete engines: graph management, parameter broadcasting, time-series bookkeeping, and skeleton run/iteration hooks.

class engine.BaseEngine[source]

Bases: object

Abstract base class for all MAIS simulation engines.

Subclasses must implement run_iteration() (and usually run()) to provide the concrete stepping logic. This class provides:

setup_model_params(model_params_dict)[source]

Broadcast scalar or list model parameters to per-node arrays.

Each value in model_params_dict is stored as an attribute of shape (num_nodes, 1). Scalars are broadcast; lists / arrays are reshaped.

Parameters:

model_params_dict (dict) – Mapping of parameter name (str) to its value (scalar, list, or numpy.ndarray).

set_seed(random_seed)[source]

Set the NumPy random seed for reproducible simulations.

Parameters:

random_seed (int) – Seed value passed to numpy.random.seed.

inicialization()[source]

Initialise model parameters and build the adjacency matrix.

Reads values from self.init_kwargs (falling back to defaults declared in fixed_model_parameters, common_arguments, and model_parameters), optionally seeds NumPy’s random number generator, calls update_graph() to build the adjacency matrix, and then calls setup_model_params() to broadcast all model parameters to per-node arrays.

setup_series_and_time_keeping()[source]

Initialise all time-tracking variables and empty time-series containers.

Sets self.t, self.tmax, self.tidx to zero and creates None-initialised containers for state_counts, state_increments, and self.N. Concrete subclasses override this method and fill the containers with proper TimeSeries objects.

states_and_counts_init(ext_nodes=None, ext_code=None)[source]

Initialise per-state node counts and the initial membership arrays.

Reads init_<STATE_LABEL> entries from self.init_kwargs to set the starting counts for each state, assigns remaining nodes to the first state, shuffles node assignments randomly, and builds the self.memberships boolean array (shape: num_states × num_nodes × 1).

Parameters:

  • ext_nodes (int, optional) – Number of external nodes to pin to ext_code at the end of the node list. Defaults to None.

  • ext_code (int, optional) – State code to assign to external nodes. Defaults to None.

Raises:

ValueError – If external nodes are not the last nodes in the list.

update_graph(new_G)[source]

Build the sparse adjacency matrix from the supplied graph object.

Accepts a scipy.sparse.csr_matrix, a numpy.ndarray, or a networkx.Graph and stores the result as self.A (CSR format). Also updates self.num_nodes and self.degree.

Parameters:

new_G – Contact network; one of scipy.sparse.csr_matrix, numpy.ndarray, or networkx.classes.graph.Graph.

Raises:

TypeError – If new_G is not a supported type.

node_degrees(Amat)[source]

Return the degree (column sum) of each node in the adjacency matrix.

Parameters:

Amat (scipy.sparse.csr_matrix) – Adjacency matrix of shape (num_nodes, num_nodes).

Returns:

Array of shape (num_nodes, 1) with each node’s degree.

Return type:

numpy.ndarray

set_periodic_update(callback)[source]

Register a callback to be invoked at the end of each simulated day.

The callback is stored as self.periodic_update_callback and is called by the engine’s run loop at midnight of each simulated day.

Parameters:

callback (callable) – Object or function to call once per day.

update_scenario_flags()[source]

Recompute boolean scenario flags (testing, tracing, etc.).

No-op in the base class. Subclasses override this to update flags such as self.testing_scenario and self.tracing_scenario based on the current parameter values.

num_contacts(state)[source]

Return the number of contacts each node has in the given state(s).

Deprecated since version Do: not use this method in newer engines; it is retained only for backward compatibility with legacy model code.

Parameters:

state (str or list of str) – State label(s) to count contacts in. A single string queries one state; a list sums over multiple states.

Returns:

Column vector of shape (num_nodes, 1) with the contact count per node.

Return type:

numpy.ndarray

Raises:

TypeException – If state is neither a str nor a list.

current_state_count(state)[source]

Return the current count of nodes in state.

Parameters:

state (int) – State code (see STATES).

Returns:

Current count, or None if the time-series has not been initialised yet.

Return type:

int or None

current_N()[source]

Return the current effective population size.

Returns:

Number of nodes not in any invisible state at the current time index.

Return type:

float

run_iteration()[source]

Advance the simulation by one step.

No-op in the base class. Subclasses implement the concrete stepping logic here and return True while the simulation should continue or False when it should stop.

Returns:

True to continue, False to stop.

Return type:

bool

run(T, print_interval=10, verbose=False)[source]

Run the simulation for T time units.

No-op in the base class. Subclasses implement the main simulation loop here.

Parameters:

  • T (int or float) – Number of time units (days) to simulate.

  • print_interval (int, optional) – Print status every this many days. Defaults to 10.

  • verbose (bool, optional) – If True, print per-state counts at each print interval. Defaults to False.

Returns:

True on successful completion.

Return type:

bool

to_df()[source]

Convert simulation output to a pandas.DataFrame.

Returns:

Indexed by simulation time step (T), with one column per state (counts) and one column per state prefixed with inc_ (increments), plus a day column.

Return type:

pandas.DataFrame

save(file_or_filename)[source]

Save the simulation time-series to a CSV file.

Calls to_df() and writes the resulting DataFrame to disk.

Parameters:

file_or_filename (str or file-like) – Destination path or open file object passed directly to pandas.DataFrame.to_csv().

save_durations(file_or_filename)[source]

Save per-state duration statistics to a file.

Not yet implemented in BaseEngine. Subclasses may override this method to write duration histograms.

Parameters:

file_or_filename (str or file-like) – Destination path or open file object.

increase_data_series_length()[source]

Extend time-series storage (no-op; handled automatically by series objects).

The underlying TimeSeries objects auto-extend, so this override is intentionally empty.

increase_history_len()[source]

Extend the transition-history buffers when storage is exhausted.

No-op in the base class (handled automatically by the series objects in most engines). Subclasses may override to call bloat() on self.tseries and self.history.

finalize_data_series()[source]

Trim all time-series to the actually consumed length (concrete implementation).

Calls finalize on self.tseries, self.history, each per-state count and increment series, self.N, and self.states_history.

print(verbose=False)[source]

Print the current simulation time and optionally per-state counts.

Parameters:

verbose (bool, optional) – If True, also print the current count for every state. Defaults to False.

engine_daily

Daily-batched Gillespie engine.

This module defines DailyEngine, which runs the Gillespie event selection intra-day but batches all state transitions to midnight, so the observable state changes once per simulated day.

class engine_daily.DailyEngine[source]

Bases: SeirsPlusLikeEngine

Gillespie engine that applies state transitions only at midnight.

Inherits from SeirsPlusLikeEngine. During the day the engine collects proposed transitions in a to-do list; at midnight (update_states() / midnight()) the transitions are committed in bulk, ensuring the observable model state changes only once per day.

inicialization()[source]

Initialise engine and allocate the daily to-do lists.

Creates empty self.todo_list and self.todo_t accumulators before delegating to the parent initialiser.

run_iteration(alpha, cumsum, transition_types)[source]

Sample the next Gillespie event and add it to the pending to-do list.

Does not apply the transition immediately; it is deferred to the next call to update_states(). At most one pending transition per node is kept (first-event-wins within a day).

Parameters:

  • alpha (float) – Total propensity (sum of all propensities).

  • cumsum (numpy.ndarray) – Cumulative-sum vector over flattened propensities, used for event selection.

  • transition_types (list) – Ordered list of (from_state, to_state) tuples matching the propensity order.

Returns:

Always True (day-level termination is handled by run()).

Return type:

bool

update_states()[source]

Commit all pending transitions accumulated during the current day.

Iterates self.todo_list and applies each transition by updating self.memberships, self.state_counts, self.history, and self.N. Clears the to-do lists afterwards.

midnight(verbose)[source]

Execute end-of-day actions: commit transitions and recalculate propensities.

Calls update_states() to apply all pending transitions, fires self.periodic_update_callback if set (updating the graph if the callback returns a new one), then recomputes propensities for the next day via propensities_recalc().

Parameters:

verbose (bool) – Passed through (currently unused).

Returns:

(alpha, cumsum, has_events, transition_types) as returned by propensities_recalc().

Return type:

tuple

print(verbose=False)[source]

Print the current simulation time and optionally per-state counts.

Parameters:

verbose (bool, optional) – If True, also print per-state counts. Defaults to False.

propensities_recalc()[source]

Recalculate propensities and return flattened cumulative-sum data.

Returns:

A 4-tuple (alpha, cumsum, has_events, transition_types) where:

  • alpha (float) – total propensity.

  • cumsum (numpy.ndarray) – cumulative sum of flattened propensity array.

  • has_events (bool) – True if total propensity is > 0.

  • transition_types (list) – ordered (from, to) transition pairs.

Return type:

tuple

run(T, print_interval=10, verbose=False)[source]

Run the daily-batched simulation for up to T time units.

Calls propensities_recalc() once at the start, then loops over run_iteration(). At each midnight calls midnight() to commit transitions and recompute propensities.

Parameters:

  • T (int or float) – Duration to simulate.

  • print_interval (int, optional) – Print status every this many days. Defaults to 10.

  • verbose (bool, optional) – If True, include per-state detail in progress messages. Defaults to False.

Returns:

True on completion, False if T <= 0.

Return type:

bool

engine_m

Multi-layer-graph sequential engine (EngineM).

This module defines EngineM, which extends SequentialEngine to work directly with a multi-layer graph object instead of a sparse adjacency matrix. Edge-level transmission probabilities are computed using the graph’s own edge-probability and edge-intensity queries, taking into account family vs. non-family contact layers and asymptomatic infectiousness reduction.

class engine_m.STATES[source]

Bases: object

State codes for EngineM (mirrors STATES).

S

Susceptible.

Type:

int

S_s

Susceptible with false symptoms.

Type:

int

E

Exposed.

Type:

int

I_n

Infectious asymptomatic (non-symptomatic track).

Type:

int

I_a

Infectious pre-symptomatic.

Type:

int

I_s

Infectious symptomatic.

Type:

int

I_ds

Infectious symptomatic, detected.

Type:

int

J_s

Post-infectious symptomatic.

Type:

int

J_n

Post-infectious asymptomatic.

Type:

int

E_d

Exposed, detected.

Type:

int

I_da

Infectious pre-symptomatic, detected.

Type:

int

I_dn

Infectious asymptomatic, detected.

Type:

int

J_ds

Post-infectious symptomatic, detected.

Type:

int

J_dn

Post-infectious asymptomatic, detected.

Type:

int

R_d

Recovered, detected.

Type:

int

R_u

Recovered, undetected.

Type:

int

D_d

Dead, detected.

Type:

int

D_u

Dead, undetected.

Type:

int

S = 0
S_s = 1
E = 2
I_n = 3
I_a = 4
I_s = 5
I_ds = 6
J_s = 11
J_n = 12
E_d = 13
I_da = 14
I_dn = 15
J_ds = 16
J_dn = 17
R_d = 7
R_u = 8
D_d = 9
D_u = 10
class engine_m.EngineM[source]

Bases: SequentialEngine

Multi-layer-graph daily-step engine (the production engine for Model-M).

Extends SequentialEngine with:

  • Multi-layer graph support via self.graph (no sparse adjacency matrix).

  • Detailed per-edge transmission computation in prob_of_contact(), accounting for edge type (family / class / pub / super-spreader), edge intensity, and asymptomatic infectiousness reduction.

This engine is used by the TGM (Task Group Model) model variant.

update_graph(new_G)[source]

Store the multi-layer graph object and update the node count.

Unlike the parent implementation, this method does not build a sparse adjacency matrix; the engine queries the graph object directly.

Parameters:

new_G – A multi-layer graph object exposing num_nodes and graph-query methods (e.g. get_edges, get_edges_probs).

node_degrees(Amat)[source]

Not implemented: EngineM uses the graph object directly.

Raises:

NotImplementedError – Always.

prob_of_contact(source_states, source_candidate_states, dest_states, dest_candidate_states, beta, beta_in_family)[source]

Compute per-node exposure probability using the multi-layer graph.

Selects active edges by flipping per-edge coins (edge probability from the graph), then determines which active edges connect nodes in the relevant source/destination states, and finally computes the compound probability that each source node is not infected (product over independent contacts) and returns 1 - P(no infection).

Distinguishes between family-type edges (using beta_in_family / self.beta_A_in_family) and other edges (using beta / self.beta_A), and reduces transmission rate for asymptomatic infectious nodes.

Parameters:

  • source_states (list of int) – States that can become exposed.

  • source_candidate_states (list of int) – Candidate source states for edge selection (superset of source_states).

  • dest_states (list of int) – Infectious states.

  • dest_candidate_states (list of int) – Candidate destination states for edge selection (superset of dest_states).

  • beta (numpy.ndarray) – Per-node baseline transmission rate for non-family contacts, shape (num_nodes, 1).

  • beta_in_family (numpy.ndarray) – Per-node baseline transmission rate for family contacts, shape (num_nodes, 1).

Returns:

Per-node exposure probability, shape (num_nodes, 1).

Return type:

numpy.ndarray

engine_seirspluslike

SEIRS-Plus-Like continuous-time (Gillespie) simulation engine.

This module defines SeirsPlusLikeEngine, which extends BaseEngine with a Gillespie-style event-driven simulation loop, network-contact propensity calculations, and contact-tracing helpers.

class engine_seirspluslike.SeirsPlusLikeEngine[source]

Bases: BaseEngine

Gillespie (continuous-time) epidemic engine for network models.

Extends BaseEngine with:

  • Propensity-based event selection (Gillespie algorithm).

  • Network-aware contact probability calculations (prob_of_contact(), num_contacts()).

  • Built-in testing / contact-tracing scenario flags (update_scenario_flags()).

  • Full time-series bookkeeping with automatic buffer extension.

Subclasses supply calc_propensities to define the concrete transition rates.

inicialization()[source]

Initialise model parameters and broadcast them to per-node arrays.

Reads constructor keyword arguments (falling back to per-parameter defaults), seeds NumPy’s RNG when a random_seed is given, calls update_graph() to build the adjacency matrix, and then broadcasts every model parameter to a (num_nodes, 1) array.

setup_series_and_time_keeping()[source]

Create all time-series buffers for a SEIRS-Plus-like run.

Allocates TimeSeries objects for state counts, state increments, population size, mean/median infection probabilities, and the transition-event log. Also initialises the contact-history ring buffer.

states_and_counts_init()[source]

Initialise per-state node counts and the initial membership array.

Reads init_<STATE_LABEL> keyword arguments, assigns residual nodes to the first state, randomly shuffles node assignments, and builds the self.memberships integer array of shape (num_states, num_nodes, 1). Also zeroes durations, infect_start, infect_time, and test_waiting arrays.

node_degrees(Amat)[source]

Return the degree of each node (sum of adjacency-matrix columns).

Parameters:

Amat (scipy.sparse.csr_matrix) – Adjacency matrix.

Returns:

Array of shape (num_nodes, 1).

Return type:

numpy.ndarray

update_scenario_flags()[source]

Recompute testing_scenario and tracing_scenario flags.

These boolean attributes are used by calc_propensities to skip expensive matrix operations when testing or tracing is inactive.

num_contacts(state)[source]

Return per-node contact counts from nodes in state.

Uses the membership arrays and the adjacency matrix to count how many neighbours of each node are currently in state.

Parameters:

state (int or list of int) – A single state code, or a list of state codes whose contacts are summed.

Returns:

Column vector of shape (num_nodes, 1).

Return type:

numpy.ndarray

Raises:

TypeException – If state is neither an int nor a list.

prob_of_contact(source_states, source_candidate_states, dest_states, dest_candidate_states, beta)[source]

Compute per-node probability of becoming newly exposed via network contacts.

Stochastically activates edges between source_candidate_states and dest_candidate_states, records active contacts, then computes for each node in source_states the probability of being exposed by at least one neighbour in dest_states.

Parameters:

  • source_states (list of int) – States that can become exposed.

  • source_candidate_states (list of int) – Candidate source states (superset of source_states) used for edge selection.

  • dest_states (list of int) – Infectious states.

  • dest_candidate_states (list of int) – Candidate destination states (superset of dest_states) used for edge selection.

  • beta (numpy.ndarray) – Per-node transmission rate, shape (num_nodes, 1).

Returns:

Per-node exposure probability, shape (num_nodes, 1).

Return type:

numpy.ndarray

current_state_count(state)[source]

Return the count of nodes in state at the current time index.

Parameters:

state (int) – State code.

Returns:

Current count.

Return type:

int

current_N()[source]

Return the current effective population size.

Returns:

Population size at the current time index (excludes invisible states).

Return type:

float

increase_data_series_length()[source]

Extend time-series and transition-history buffers.

Called automatically by run_iteration() when storage is nearly exhausted. Calls bloat() on all relevant TimeSeries objects.

finalize_data_series()[source]

Trim all time-series to the actually consumed length.

Called at the end of a run. Invokes finalize(self.tidx) on self.tseries, self.history, all state-count / state-increment series, and self.N.

save(file_or_filename)[source]

Save the simulation time-series to a CSV file.

Assembles per-state counts and increments, saves them using the floating-point event time as the index.

Parameters:

file_or_filename (str or file-like) – Destination path or open file object.

run_iteration()[source]

Perform one Gillespie step: select and apply one state transition.

Uses the propensities returned by calc_propensities to sample the next event time (exponential distribution) and the next transition (multinomial selection). Updates memberships, state-count time-series, and the transition history.

Returns:

True if the simulation should continue; False if the maximum time has been reached or no further transitions are possible.

Return type:

bool

run(T, print_interval=10, verbose=False)[source]

Run the Gillespie simulation for up to T time units.

Loops over run_iteration() until the simulation terminates or T is exhausted. Fires self.periodic_update_callback at each midnight and optionally prints progress.

Parameters:

  • T (int or float) – Number of time units to simulate.

  • print_interval (int, optional) – Print status every this many simulated days. Set to 0 or negative to suppress all output. Defaults to 10.

  • verbose (bool, optional) – If True, include per-state counts in progress output. Defaults to False.

Returns:

True on successful completion, False if T <= 0.

Return type:

bool

engine_sequential

Sequential (discrete-time daily-step) epidemic simulation engine.

This module defines SequentialEngine, which overrides the continuous- time Gillespie loop of SeirsPlusLikeEngine with a discrete daily-step update: all nodes draw new states simultaneously once per day using roulette-wheel selection over the propensity matrix.

It also provides an extended STATES enumeration and the helper _searchsorted2d() used for vectorised roulette-wheel selection.

class engine_sequential.STATES[source]

Bases: object

Extended integer state codes used by SequentialEngine models.

S

Susceptible.

Type:

int

S_s

Susceptible with false symptoms.

Type:

int

E

Exposed.

Type:

int

I_n

Infectious, asymptomatic non-symptomatic track.

Type:

int

I_a

Infectious, pre-symptomatic.

Type:

int

I_s

Infectious, symptomatic.

Type:

int

I_ds

Infectious, symptomatic, detected.

Type:

int

J_s

Post-infectious, symptomatic.

Type:

int

J_n

Post-infectious, asymptomatic.

Type:

int

E_d

Exposed, detected.

Type:

int

I_da

Infectious, pre-symptomatic, detected.

Type:

int

I_dn

Infectious, asymptomatic, detected.

Type:

int

J_ds

Post-infectious, symptomatic, detected.

Type:

int

J_dn

Post-infectious, asymptomatic, detected.

Type:

int

R_d

Recovered, detected.

Type:

int

R_u

Recovered, undetected.

Type:

int

D_d

Dead, detected.

Type:

int

D_u

Dead, undetected.

Type:

int

S = 0
S_s = 1
E = 2
I_n = 3
I_a = 4
I_s = 5
I_ds = 6
J_s = 11
J_n = 12
E_d = 13
I_da = 14
I_dn = 15
J_ds = 16
J_dn = 17
R_d = 7
R_u = 8
D_d = 9
D_u = 10
class engine_sequential.SequentialEngine[source]

Bases: SeirsPlusLikeEngine

Discrete daily-step engine (roulette-wheel selection over propensities).

Replaces the continuous-time Gillespie loop of SeirsPlusLikeEngine with a synchronous, per-node roulette-wheel update executed once per simulated day. All propensities must therefore sum to 1 across transitions for each node (i.e. they are interpreted as probabilities rather than rates).

Additional public methods allow manual state manipulation for scenario modelling: move_to_E(), move_to_R(), force_infect(), detected_node().

inicialization()[source]

Initialise the sequential engine and allocate the testable array.

Calls the parent inicialization(), then creates self.testable — a per-node boolean array tracking whether a node will seek a test when symptomatic.

run_iteration()[source]

Perform one day of simulation: compute propensities and update states.

For every node draws a uniform random number and selects its next state via roulette-wheel selection over the column of propensities. State changes are accumulated in self.delta and applied in a single batch at the end of the day.

Returns:

Always True (termination is managed by run()).

Return type:

bool

run(T, print_interval=0, verbose=False)[source]

Run the simulation for T days with daily-step updates.

Iterates over days 1..T calling run_iteration() each day. Fires self.periodic_update_callback when set. If the epidemic ends early (all unstable counts zero), fills remaining days with the last observed counts.

Parameters:

  • T (int) – Number of days to simulate.

  • print_interval (int, optional) – Print status every this many days. Set to 0 to suppress. Defaults to 0.

  • verbose (bool, optional) – If True, print per-state counts at each interval. Defaults to False.

Returns:

Always True.

Return type:

bool

increase_data_series_length()[source]

Extend all daily time-series buffers by 100 entries.

Called automatically when the pre-allocated storage is exhausted.

increase_history_len()[source]

Extend the event-history and time-series buffers.

Enlarges self.tseries and self.history by 10 * num_nodes entries.

finalize_data_series()[source]

Trim all time-series to the actually consumed length.

Calls finalize(self.t) on every daily TimeSeries and finalize(self.tidx) on the event-log series.

current_state_count(state)[source]

Return the count of nodes in state at the current day.

Overrides the parent method: uses self.t (day index) rather than self.tidx (event index).

Parameters:

state (int) – State code.

Returns:

Node count on the current day.

Return type:

int

current_N()[source]

Return the effective population size on the current day.

Overrides the parent method: uses self.t rather than self.tidx.

Returns:

Population size (excluding invisible-state nodes).

Return type:

float

get_state_count(state=None)[source]

Return per-day count time-series for a given state or all states.

Parameters:

state (int, optional) – State code. If None, returns the full state_counts dictionary. Defaults to None.

Returns:

The per-day count series for state, or the complete state_counts mapping when state is None.

Return type:

TimeSeries or dict

to_df()[source]

Convert simulation output to a pandas.DataFrame.

Extends the parent to_df() with additional test-related columns: tests, quarantine_tests, sum_of_waiting, and all_positive_tests.

Returns:

Combined state-count, increment, and test statistics indexed by day.

Return type:

pandas.DataFrame

save_durations(f)[source]

Write per-state duration lists to an open file as CSV rows.

Each row contains the state label followed by comma-separated integer durations (in days).

Parameters:

f (file-like) – Open writable file object.

save_node_states(filename)[source]

Write the per-node state history to a CSV file.

Parameters:

filename (str) – Destination file path.

move_to_E(num)[source]

Randomly expose num susceptible nodes by moving them to state E.

Only nodes currently in S or S_s are eligible. The operation updates state counts, increments, membership arrays, and duration tracking for the current day.

Parameters:

num (int) – Number of nodes to expose.

move_to_R(nodes)[source]

Move a specific set of exposed nodes directly to the detected-recovered state.

Only nodes currently in state E are supported.

Parameters:

nodes (iterable of int) – Node indices to move.

Raises:

ValueError – If any node in nodes is not currently in state E.

force_infect(nodes)[source]

Forcibly move a set of nodes to the symptomatic-infectious state I_s.

Dead nodes (D_d, D_u) are silently skipped. All other nodes are moved regardless of their current state. Intended for scenario seeding.

Parameters:

nodes (iterable of int) – Node indices to infect.

detected_node(node_number)[source]

Mark a node as detected (positive test) and transition it accordingly.

Maps the node’s current undetected state to its detected counterpart (e.g. E → E_d, I_s → I_ds). If the node is already in a detected or terminal state, the call is a no-op. Also updates test-waiting statistics.

Parameters:

node_number (int) – Index of the node that tested positive.

Raises:

ValueError – If the node is in an unexpected state.

extended_network_model

Extended SEIRS network model definition.

This module defines the full epidemiological compartment model used in the MAIS project (ExtendedNetworkModel and its variants). It contains:

  • STATES – integer compartment codes and the detected subset.

  • state_codes – human-readable label mapping.

  • model_definition – the complete model specification dict (states, transitions, parameters).

  • calc_propensities() – per-node daily transition-probability function.

  • Four model classes created via create_custom_model():

    • ExtendedNetworkModel – Gillespie (continuous-time) engine.

    • ExtendedDailyNetworkModel – daily-batched Gillespie engine.

    • ExtendedSequentialNetworkModel – discrete-step sequential engine.

    • TGMNetworkModel – multi-layer-graph engine (EngineM).

class extended_network_model.STATES[source]

Bases: object

Extended state codes including detected variants of every compartment.

S

Susceptible.

Type:

int

S_s

Susceptible with false symptoms.

Type:

int

E

Exposed.

Type:

int

I_n

Infectious, asymptomatic non-symptomatic track.

Type:

int

I_a

Infectious, pre-symptomatic.

Type:

int

I_s

Infectious, symptomatic.

Type:

int

I_ds

Infectious, symptomatic, detected.

Type:

int

J_s

Post-infectious, symptomatic.

Type:

int

J_n

Post-infectious, asymptomatic.

Type:

int

E_d

Exposed, detected.

Type:

int

I_da

Infectious, pre-symptomatic, detected.

Type:

int

I_dn

Infectious, asymptomatic, detected.

Type:

int

J_ds

Post-infectious, symptomatic, detected.

Type:

int

J_dn

Post-infectious, asymptomatic, detected.

Type:

int

R_d

Recovered, detected.

Type:

int

R_u

Recovered, undetected.

Type:

int

D_d

Dead, detected.

Type:

int

D_u

Dead, undetected.

Type:

int

detected

Subset of state codes that are in a detected state.

Type:

set

S = 0
S_s = 1
E = 2
I_n = 3
I_a = 4
I_s = 5
I_ds = 6
J_s = 11
J_n = 12
E_d = 13
I_da = 14
I_dn = 15
J_ds = 16
J_dn = 17
R_d = 7
R_u = 8
D_d = 9
D_u = 10
detected = {6, 13, 14, 15, 16, 17}
extended_network_model.calc_propensities(model, use_dict=True)[source]

Compute per-node daily transition probabilities for the extended model.

Returns a list of propensity arrays (one per transition) ordered to match model.transitions. Each array has shape (num_nodes, 1) with values in [0, 1] representing the probability that each node undergoes the corresponding transition on the current day.

The function:

  1. Calls model.prob_of_contact to obtain the per-node infection probability P1 from network contacts, and combines it with a mass-action term P2 (controlled by model.p).

  2. Computes propensities for every state → state transition defined in model.transitions, taking into account testing rates, detection probabilities, symptom development, and mortality.

  3. Ensures each node’s propensities sum to 1 (clip to [0, 1]).

Parameters:

  • model – Active simulation-model instance with all parameters and membership arrays set.

  • use_dict (bool, optional) – Unused parameter retained for API compatibility. Defaults to True.

Returns:

One propensity array per transition, each of shape (num_nodes, 1).

Return type:

list of numpy.ndarray

load_model

model

Factory for creating custom epidemic network-model classes at runtime.

This module provides create_custom_model(), which dynamically builds a Python class that combines a user-supplied model definition (states, transitions, parameters, propensity function) with a chosen simulation engine (default: SeirsPlusLikeEngine).

model.not_implemented_yet()[source]

Placeholder that raises NotImplementedError.

Used as a sentinel default for calc_propensities when none is supplied to create_custom_model().

Raises:

NotImplementedError – Always.

model.create_custom_model(clsname, states, state_str_dict, transitions, final_states=[], invisible_states=[], unstable_states=[], init_arguments={}, model_parameters={}, calc_propensities=<function not_implemented_yet>, member_functions=None, engine=<class 'models.engine_seirspluslike.SeirsPlusLikeEngine'>)[source]

Dynamically create an epidemic model class from a model definition.

Builds and returns a new Python class named clsname that inherits from engine and is populated with the supplied model metadata, parameters, and optional member functions. The generated class receives an __init__ that accepts the contact graph G and keyword arguments corresponding to all declared parameters.

Parameters:

  • clsname (str) – Name of the class to create.

  • states (list) – Ordered list of integer state codes.

  • state_str_dict (dict) – Mapping from state code (int) to label (str).

  • transitions (list) – List of (from_state, to_state) tuples that describe all allowed state transitions.

  • final_states (list, optional) – States from which no further transition occurs (absorbing states). Defaults to [].

  • invisible_states (list, optional) – States whose members are excluded from the active population count (e.g. deceased). Defaults to [].

  • unstable_states (list, optional) – States that can still change; used to decide when to terminate a simulation run. Defaults to [], which is treated as all states.

  • init_arguments (dict, optional) – Fixed constructor arguments as {name: (default, description)}. Defaults to {}.

  • model_parameters (dict, optional) – Per-node model parameters as {name: (default, description)}. Values may be scalars or arrays of length num_nodes. Defaults to {}.

  • calc_propensities (callable, optional) – Function with signature calc_propensities(model) that returns a list of per-node propensity arrays. Used by SEIRS-Plus-Like derived engines. Defaults to not_implemented_yet().

  • member_functions (dict, optional) – Additional methods to attach to the generated class, as {method_name: function}. Defaults to None.

  • engine (type, optional) – Base engine class. Defaults to SeirsPlusLikeEngine.

Returns:

A new class inheriting from engine with all model metadata set as class variables and a suitable __init__ attached.

Return type:

type

model_zoo

Registry of all available simulation model classes.

model_zoo is a dictionary mapping model-name strings to their corresponding class objects. It is used by loader utilities (e.g. load_model_from_config) to look up the correct class by name at runtime.

Available models:

  • "ExtendedNetworkModel" – ExtendedNetworkModel (Gillespie engine).

  • "ExtendedDailyNetworkModel" – ExtendedDailyNetworkModel (daily Gillespie engine).

  • "ExtendedSequentialNetworkModel" – ExtendedSequentialNetworkModel (sequential discrete-step engine).

  • "TGMNetworkModel" – TGMNetworkModel (multi-layer-graph engine).

  • "SimulationDrivenModel" – SimulationDrivenModel (agent- based plan engine; currently the primary supported model).

  • "InfoSIRModel" – InfoSIRModel.

  • "InfoTippingModel" – InfoTippingModel.

  • "RumourModel" – RumourModel.

  • "RumourModelInfo" – RumourModelInfo.

prob_infection

Network contact and infection probability utilities.

This module provides standalone functions that compute which edges become “active” on a given day and whether susceptible nodes are exposed via those active edges. Two main entry points are provided:

Helper functions select_active_edges(), archive_active_edges(), and get_relevant_edges() are used internally by both entry points.

prob_infection.select_active_edges(model, source_states, source_candidate_states, dest_states, dest_candidate_states)[source]

Stochastically select active (contact) edges between two node sets.

For each edge connecting a node in source_candidate_states to a node in dest_candidate_states, flips a biased coin with the edge’s daily-contact probability. Returns only edges where contact actually occurred.

Parameters:

  • model – The active simulation model instance (must expose memberships, graph, num_nodes).

  • source_states (list of int) – States whose nodes can become exposed (used for final filtering, not edge selection).

  • source_candidate_states (list of int) – Superset of source_states used for initial candidate-edge selection.

  • dest_states (list of int) – Infectious states (used for filtering).

  • dest_candidate_states (list of int) – Superset of dest_states used for initial candidate-edge selection.

Returns:

(active_edges, active_edges_dirs) where both are numpy.ndarray of shape (num_active_edges,), or (None, None) if no edges are active.

Return type:

tuple

prob_infection.archive_active_edges(model, active_edges, active_edges_dirs)[source]

Record today’s active edges in the model’s contact-history buffer.

Retrieves source and destination node indices for each active edge and appends them (along with edge type) to model.contact_history.

Parameters:

  • model – The active simulation model instance.

  • active_edges (numpy.ndarray) – Indices of today’s active edges.

  • active_edges_dirs (numpy.ndarray) – Direction flags for each edge (True = canonical direction, False = reversed).

prob_infection.get_relevant_edges(model, active_edges, active_edges_dirs, source_states, dest_states)[source]

Filter active edges to those connecting relevant source/destination states.

From the set of active_edges retains only those where the source node is in one of source_states and the destination node is in one of dest_states.

Parameters:

  • model – The active simulation model instance.

  • active_edges (numpy.ndarray) – Indices of today’s active edges.

  • active_edges_dirs (numpy.ndarray) – Direction flags for active_edges.

  • source_states (list of int) – States eligible to be exposed.

  • dest_states (list of int) – Infectious states.

Returns:

(active_relevant_edges, active_relevant_edges_dirs) or (None, None) if no relevant edges remain after filtering.

Return type:

tuple

prob_infection.prob_of_contact_old(model, source_states, source_candidate_states, dest_states, dest_candidate_states, beta, beta_in_family)[source]

Legacy implementation of the contact-probability computation (abandoned).

Retained for historical reference only. Prefer prob_of_contact().

Parameters:

  • model – The active simulation model instance.

  • source_states (list of int) – States that can be exposed.

  • source_candidate_states (list of int) – Candidate source states for edge selection.

  • dest_states (list of int) – Infectious states.

  • dest_candidate_states (list of int) – Candidate destination states for edge selection.

  • beta (numpy.ndarray) – Per-node non-family transmission rate, shape (num_nodes, 1).

  • beta_in_family (numpy.ndarray) – Per-node family transmission rate, shape (num_nodes, 1).

Returns:

Per-node binary exposure indicator, shape (num_nodes, 1).

Return type:

numpy.ndarray

prob_infection.prob_of_contact(model, source_states, source_candidate_states, dest_states, dest_candidate_states, beta, beta_in_family)[source]

Evaluate per-node exposure via stochastic edge activation (production).

For each graph edge, independently activates it with the edge’s daily contact probability. For active edges connecting a susceptible source to an infectious destination, draws a Bernoulli trial with the effective transmission rate (accounting for family vs. non-family edges and asymptomatic reduction). Returns a binary indicator vector: 1 for each node that was exposed on this day.

source_candidate_states and dest_candidate_states are accepted for backward compatibility but are no longer used; the function queries all edges.

Updates the following model statistics:

  • model.contact_history – active edge contacts.

  • model.successfull_source_of_infection – per-node successful infection counts.

  • model.stat_successfull_layers – per-layer successful transmission counts.

Parameters:

  • model – The active simulation model instance.

  • source_states (list of int) – States that can become exposed.

  • source_candidate_states (list of int) – Unused (backward compat.).

  • dest_states (list of int) – Infectious states.

  • dest_candidate_states (list of int) – Unused (backward compat.).

  • beta (numpy.ndarray) – Per-node non-family transmission rate, shape (num_nodes, 1).

  • beta_in_family (numpy.ndarray) – Per-node family transmission rate, shape (num_nodes, 1).

Returns:

Per-node binary exposure indicator, shape (num_nodes, 1). Value is 1 if the node was newly exposed, 0 otherwise.

Return type:

numpy.ndarray

simulation_engine

Plan-based discrete-time simulation engine.

This module defines SimulationEngine, which replaces the Gillespie propensity loop with a planning approach: each node holds a pre-scheduled (time_to_go, state_to_go) pair. Every day the engine decrements counters and moves nodes whose countdown has reached zero, then invokes daily_update() for state-dependent daily checks (e.g. infection attempts).

class simulation_engine.SimulationEngine(G, **kwargs)[source]

Bases: BaseEngine

Discrete-time plan-based epidemic engine.

Each simulated agent (node) carries a plan: a countdown (time_to_go) and a target state (state_to_go). On every simulated day:

  1. daily_update() is called for nodes that need a check (e.g. susceptible nodes that might be infected).

  2. All countdown timers are decremented.

  3. Nodes whose timer hits zero are moved to their planned state via change_states().

  4. update_plan() sets new plans for the nodes that just moved.

Subclasses implement daily_update() and update_plan() to define the model’s disease-progression logic.

Class-level attributes (override in subclasses):

states (list): Ordered list of state codes. num_states (int): Number of states. state_str_dict (dict): State-code → label mapping. ext_code (int): State code used for external nodes. transitions (list): Allowed (from, to) pairs. num_transitions (int): Number of transitions. final_states (list): Absorbing states. invisible_states (list): States excluded from population count. unstable_states (list): States that can still change. fixed_model_parameters (dict): Scalar constructor parameters. model_parameters (dict): Per-node constructor parameters. common_arguments (dict): Common constructor parameters (seed, etc.).

states = []
num_states = 0
state_str_dict = {}
ext_code = 0
transitions = []
num_transitions = 0
final_states = []
invisible_states = []
unstable_states = []
fixed_model_parameters = {}
model_parameters = {}
common_arguments = {'random_seed': (None, 'random seed value'), 'start_day': (1, 'day to start')}
__init__(G, **kwargs)[source]

Initialise the simulation engine on a contact graph.

Parameters:

  • G – Contact graph or multi-layer graph object. Stored as both self.G (backward compatibility) and self.graph.

  • **kwargs – Keyword arguments that override any default declared in fixed_model_parameters, model_parameters, or common_arguments. State initial counts supplied as init_<STATE_LABEL>=<count>.

update_graph(new_G)[source]

Update the internal graph reference and derived node metadata.

Safe to call with None (no-op). Updates self.graph, self.num_nodes, self.num_ext_nodes, and self.nodes.

Parameters:

new_G – New graph object, or None to leave the graph unchanged.

inicialization()[source]

Initialise model parameters and build node-index array.

Delegates to the parent inicialization(), then stores a (num_nodes, 1) array of node indices in self.nodes and caches self.num_nodes.

setup_series_and_time_keeping()[source]

Create time-series buffers and per-node tracking arrays.

Extends the parent setup with:

  • Event-log buffers (tseries, history).

  • State-history array (size depends on global_configs.SAVE_NODES).

  • Per-state duration lists (when global_configs.SAVE_DURATIONS).

  • Per-node durations counter.

  • Per-state TimeSeries for counts and increments (pre-allocated to EXPECTED_NUM_DAYS entries).

states_and_counts_init(ext_nodes=None, ext_code=None)[source]

Initialise state counts and per-node planning arrays.

Extends the parent states_and_counts_init() with:

  • self.time_to_go – per-node countdown to next transition (-1 means “no scheduled transition”).

  • self.state_to_go – planned next state for each node.

  • self.current_state – copy of the initial state assignment.

  • self.need_update – boolean flag per node indicating that the plan must be recomputed.

Parameters:

  • ext_nodes (int, optional) – Number of external nodes. Defaults to None.

  • ext_code (int, optional) – State code for external nodes. Defaults to None.

daily_update(nodes)[source]

Perform daily per-node checks (e.g. infection attempts).

Called once per day for nodes flagged in self.need_check. No-op in the base class. Subclasses override this to implement infection logic and other daily events.

Parameters:

nodes (numpy.ndarray) – Boolean bitmap of nodes that require a daily check.

change_states(nodes, target_state=None)[source]

Move nodes to their planned (or a forced) target state.

Clears the old membership, assigns the new state, updates state_counts, state_increments, and optionally states_history, then calls update_plan() so each node gets a fresh plan.

Parameters:

  • nodes (numpy.ndarray) – Boolean bitmap indicating which nodes should change state.

  • target_state (int, optional) – If given, all nodes are moved to this state, ignoring self.state_to_go. If None (default), each node is moved to its own self.state_to_go value.

update_plan(nodes)[source]

Generate new transition plans for nodes that just changed state.

Sets self.time_to_go and self.state_to_go for each node in nodes based on the node’s current state. No-op in the base class. Subclasses override this to implement state-specific duration sampling.

Parameters:

nodes (numpy.ndarray) – Boolean bitmap of nodes whose plans need updating.

print(verbose=False)[source]

Print the current calendar day and optionally per-state counts.

Parameters:

verbose (bool, optional) – If True, prints T (calendar day) and the count for every state. Defaults to False.

save_durations(f)[source]

Write per-state duration lists to an open file as CSV rows.

Parameters:

f (file-like) – Open writable file object.

save_node_states(filename)[source]

Save the per-node daily state history to a CSV file.

If global_configs.SAVE_NODES is False, logs a warning and returns an empty DataFrame.

Parameters:

filename (str) – Destination file path.

Returns:

Empty DataFrame when node states were not saved.

Return type:

pandas.DataFrame

to_df()[source]

Convert simulation output to a pandas.DataFrame.

Extends the parent to_df() by adjusting the day column and the index when start_day is not 1.

Returns:

State-count and increment time-series with calendar-day index.

Return type:

pandas.DataFrame

run(T, print_interval=10, verbose=False)[source]

Run the plan-based simulation for T days.

Iterates over days 1..T, calling run_iteration() each day and the periodic callback when set. If the epidemic ends before T days, fills remaining days with the last observed counts.

Parameters:

  • T (int) – Number of days to simulate.

  • print_interval (int, optional) – Print status every this many days; 0 or negative suppresses output. Defaults to 10.

  • verbose (bool, optional) – If True, include per-state detail. Defaults to False.

Returns:

Always True.

Return type:

bool

run_iteration()[source]

Perform one day of plan-based simulation.

Steps performed each day:

  1. Copies previous-day state counts and resets increments.

  2. Increments all duration counters.

  3. Calls daily_update() for nodes that need a check.

  4. Decrements self.time_to_go for all nodes.

  5. Moves nodes whose countdown reached zero via change_states().

  6. Saves duration statistics when configured.

states

Epidemic state definitions for the agent-based network model.

This module defines the integer codes used to identify each epidemiological compartment and provides a human-readable mapping via state_codes.

class states.STATES[source]

Bases: object

Integer constants representing each epidemiological compartment.

S

Susceptible.

Type:

int

S_s

Susceptible with false symptoms.

Type:

int

E

Exposed (latent infection).

Type:

int

I_n

Infectious, asymptomatic (non-symptomatic track).

Type:

int

I_a

Infectious, pre-symptomatic (asymptomatic phase before symptoms develop).

Type:

int

I_s

Infectious, symptomatic.

Type:

int

J_s

Post-infectious, symptomatic (RNA-positive, symptomatic).

Type:

int

J_n

Post-infectious, asymptomatic (RNA-positive, no symptoms).

Type:

int

R

Recovered.

Type:

int

D

Dead.

Type:

int

EXT

External node (not counted in the population).

Type:

int

S = 0
S_s = 1
E = 2
I_n = 3
I_a = 4
I_s = 5
J_s = 6
J_n = 7
R = 8
D = 9
EXT = 10

src/policies

contact_tracing

customised_policy

Highly configurable custom policy for the MAIS epidemic simulation.

This module provides CustomPolicy, the primary policy used in production simulations. It orchestrates:

  • Layer-weight changes read from a scenario calendar file.

  • Model parameter changes (beta, theta, test rate).

  • Face-mask effectiveness updates.

  • Superspreader event toggling.

  • Forced infections on a specified day.

  • Daily background import of exposed individuals.

  • Dynamic start/stop of sub-policies loaded by name.

class customised_policy.CustomPolicy(graph, model, layer_changes_filename=None, param_changes_filename=None, policy_calendar_filename=None, beta_factor_filename=None, face_masks_filename=None, theta_filename=None, test_rate_filename=None, superspreader_date=None, superspreader_layer=None, force_infect=None, force_infect_layer=None, init_filename=None, reduction_coef1=None, reduction_coef2=None, new_beta=None, daily_import=None, **kwargs)[source]

Bases: Policy

Highly configurable orchestration policy for MAIS simulations.

Controls layer weights, model parameters (beta, theta, test rate), face-mask effectiveness, superspreader events, forced infections, background import of exposed individuals, and the lifecycle of dynamically loaded sub-policies.

All time-indexed inputs are read from external files (JSON or CSV) keyed on integer simulation-day numbers.

Parameters:

  • graph – The contact network graph object.

  • model – The epidemic model instance.

  • layer_changes_filename (str, optional) – Path to a scenario calendar file for layer-weight updates.

  • param_changes_filename (str, optional) – Temporarily disabled. Path to a JSON file of model-parameter changes.

  • policy_calendar_filename (str, optional) – Path to a JSON file mapping simulation days to [action, policy_string] pairs for starting/stopping sub-policies.

  • beta_factor_filename (str, optional) – Path to a JSON/CSV file with per-day beta-factor multipliers.

  • face_masks_filename (str, optional) – Path to a JSON/CSV file with per-day face-mask compliance values.

  • theta_filename (str, optional) – Path to a JSON/CSV file with per-day theta_Is multipliers.

  • test_rate_filename (str, optional) – Path to a JSON/CSV file with per-day test-rate multipliers.

  • superspreader_date (int or str, optional) – Simulation day on which the superspreader layer is activated (one day only).

  • superspreader_layer (int or str, optional) – Index of the superspreader layer (default 31).

  • force_infect (int or str, optional) – Simulation day on which one node on force_infect_layer is forcibly infected.

  • force_infect_layer (int or str, optional) – Layer index used to select the node for forced infection.

  • init_filename (str, optional) – Path to a JSON file mapping days to the number of nodes moved to state E on that day.

  • reduction_coef1 (float or str) – Required. Primary beta reduction coefficient (face-mask effect on non-family contacts).

  • reduction_coef2 (float or str) – Required. Secondary beta reduction coefficient (face-mask spillover into family).

  • new_beta (str, optional) – "Yes" to use the updated beta calculation combining face-mask compliance and beta factors.

  • daily_import (float or str, optional) – Daily probability of importing one exposed individual (value in [0, 1]).

  • **kwargs – Sub-policy keyword arguments. Pass sub_policies (list of names) together with <name>_filename, <name>_name, and optionally <name>_config for each sub-policy.

Raises:

ValueError – If param_changes_filename is provided (temporarily disabled), if an unknown action is found in the policy calendar, if new_beta=Yes but the required calendars are absent or mismatched, or if reduction_coef1/ reduction_coef2 are missing.

__init__(graph, model, layer_changes_filename=None, param_changes_filename=None, policy_calendar_filename=None, beta_factor_filename=None, face_masks_filename=None, theta_filename=None, test_rate_filename=None, superspreader_date=None, superspreader_layer=None, force_infect=None, force_infect_layer=None, init_filename=None, reduction_coef1=None, reduction_coef2=None, new_beta=None, daily_import=None, **kwargs)[source]

Initialise the custom policy from file-based calendars and keyword arguments.

Parameters:

  • graph – The contact network graph object.

  • model – The epidemic model instance.

  • layer_changes_filename (str, optional) – Layer-weight scenario calendar file.

  • param_changes_filename (str, optional) – Temporarily disabled model-parameter changes file.

  • policy_calendar_filename (str, optional) – Sub-policy lifecycle calendar file.

  • beta_factor_filename (str, optional) – Beta-factor calendar file.

  • face_masks_filename (str, optional) – Face-mask compliance calendar file.

  • theta_filename (str, optional) – theta_Is multiplier calendar file.

  • test_rate_filename (str, optional) – Test-rate multiplier calendar file.

  • superspreader_date (int or str, optional) – Day to activate superspreader layer.

  • superspreader_layer (int or str, optional) – Superspreader layer index.

  • force_infect (int or str, optional) – Day for forced infection.

  • force_infect_layer (int or str, optional) – Layer for selecting the force-infected node.

  • init_filename (str, optional) – Initial exposure calendar file.

  • reduction_coef1 (float or str) – Primary beta reduction coefficient (required).

  • reduction_coef2 (float or str) – Secondary beta reduction coefficient (required).

  • new_beta (str, optional) – "Yes" to use new beta calculation.

  • daily_import (float or str, optional) – Daily import probability.

  • **kwargs – Sub-policy configuration keyword arguments.

create_policy(filename, object_name, config_file=None)[source]

Dynamically import and instantiate a policy class by module and name.

Parameters:

  • filename (str) – Module name within the policies package (e.g. "contact_tracing").

  • object_name (str) – Class name within that module.

  • config_file (str, optional) – Path to a configuration file to pass to the policy constructor.

Returns:

An instantiated policy object.

Return type:

Policy

update_layers(coefs)[source]

Update graph layer weights to the provided coefficient values.

Parameters:

coefs – Iterable of new layer-weight values passed directly to graph.set_layer_weights.

switch_on_superspread()[source]

Activate the superspreader layer by setting its weight to 1.0.

switch_off_superspread()[source]

Deactivate the superspreader layer by setting its weight to 0.0.

update_beta(masks)[source]

Update model beta values based on face-mask compliance (legacy method).

Scales non-family beta by (1 - reduction_coef1 * masks) and family beta by a secondary factor derived from the non-family reduction.

Parameters:

masks (float) – Current face-mask compliance level in [0, 1].

update_beta2(masks, beta_factors=None)[source]

Update model beta values using both face-mask compliance and beta factors.

Family beta is reduced by (1 - reduction_coef1 * beta_factors). Non-family beta is then derived as (1 - reduction_coef2 * masks) * family_beta.

Parameters:

  • masks (float) – Current face-mask compliance in [0, 1].

  • beta_factors (float) – Additional beta scaling factor.

Raises:

AssertionError – If beta_factors is None.

beta_increase()[source]

Increase all beta values by a factor of 1.5 (mutation simulation).

Sets mutation_coef to 1.5 and scales beta, beta_A, beta_A_in_family, and beta_in_family accordingly.

update_test_rate(coef)[source]

Scale the testing rate (theta_Is) by a given coefficient.

Parameters:

coef (float) – Multiplier applied to the original test rate from model.init_kwargs.

update_theta(coef)[source]

Scale the symptomatic testing probability (theta_Is) by a coefficient.

Parameters:

coef (float) – Multiplier applied to the original theta_Is value from model.init_kwargs.

run()[source]

Execute one time-step of the custom policy.

Processes all registered calendars in order: initial exposures, daily import, policy-calendar events, parameter changes, layer updates, forced infections, superspreader toggling, face-mask updates, theta updates, test-rate updates, and finally runs all active sub-policies.

to_df()[source]

Merge and return DataFrames from all active sub-policies.

Returns:

Outer-merged DataFrame indexed by T combining statistics from all sub-policies, or None if there are no sub-policies or none produce data.

Return type:

pandas.DataFrame or None

depo

Deposit (Depo) utility for tracking nodes held for a fixed duration.

This module provides the Depo class used by policy objects to place nodes into quarantine or isolation for a configurable number of time-steps and to release them automatically when their sentence expires.

class depo.Depo(size)[source]

Bases: object

Deposit object for holding nodes for a fixed number of time-steps.

Used in policies to store nodes (e.g. quarantined individuals) for a given period of time. Internally maintains a counter array; each entry holds the remaining number of time-steps before that node is released. A value of zero means the node is not currently held.

Parameters:

size (int) – Total number of nodes in the simulation (length of the internal counter array).

__init__(size)[source]

Initialise the deposit with a counter array of zeros.

Parameters:

size (int) – Number of nodes (length of the counter array).

property num_of_prisoners

Return the number of nodes currently held in the deposit.

Returns:

Count of nodes whose remaining time is greater than zero.

Return type:

int

lock_up(nodes, duration=14, check_duplicate=False)[source]

Place nodes into the deposit for a given duration.

Parameters:

  • nodes (list or numpy.ndarray) – Node indices to lock up.

  • duration (int or numpy.ndarray) – Number of time-steps each node should remain in the deposit. May be a scalar applied to all nodes or an array of per-node values. Defaults to 14.

  • check_duplicate (bool) – If True, nodes that are already in the deposit are silently ignored (their remaining time is not updated). If False (default), an assertion error is raised if any node is already held.

Raises:

AssertionError – If nodes is not a list or numpy.ndarray, or if check_duplicate is False and any node is already in the deposit.

filter_locked(candidates)[source]

Return the subset of candidates that are not yet in the deposit.

Parameters:

candidates (numpy.ndarray) – Array of node indices to filter.

Returns:

Subset of candidates whose deposit counter is currently zero (i.e. not locked up).

Return type:

numpy.ndarray

filter_locked_bitmap(candidates)[source]

Return a boolean mask indicating which candidates are not in the deposit.

Parameters:

candidates (numpy.ndarray) – Array of node indices to check.

Returns:

Boolean array of the same length as candidates. Entry is True if the corresponding candidate is not currently in the deposit, False if it is already locked up.

Return type:

numpy.ndarray

tick_and_get_released()[source]

Advance the deposit by one time-step and return newly released nodes.

Decrements all non-zero counters by one. Nodes whose counter reaches zero are considered released.

Returns:

Array of node indices that were released (i.e. had a counter of exactly 1 before the decrement).

Return type:

numpy.ndarray

is_locked(node_id)[source]

Check whether a single node is currently held in the deposit.

Parameters:

node_id (int) – Index of the node to check.

Returns:

True if the node’s remaining time is greater than zero, False otherwise.

Return type:

bool

eva_policy

info_spreader

Information-spreader seeding policy for the MAIS simulation.

This module defines the Spreader policy, which uses PageRank centrality to seed a highly-connected node into state I at the start of the simulation. It is intended for use with information- diffusion models rather than epidemic models.

class info_spreader.Spreader(graph, model, quantile=0.9)[source]

Bases: Policy

Policy that seeds the most central node into the infectious state.

On the first simulation day, a graph-tool graph is built from the contact network, PageRank centrality is computed (weighted by edge probability, intensity, and layer weight), and the node at the requested quantile of centrality is moved to state I.

Parameters:

  • graph – The contact network graph object. Must expose e_source, e_dest, e_probs, e_intensities, e_types, layer_weights, and num_nodes.

  • model – The epidemic model instance (must support change_states and nodes).

  • quantile (float) – Quantile in [0, 1] of the centrality distribution used to select the seed node. Defaults to 0.9.

__init__(graph, model, quantile=0.9)[source]

Initialise the spreader policy.

Parameters:

  • graph – The contact network graph object.

  • model – The epidemic model instance.

  • quantile (float) – Centrality quantile for seed node selection. Defaults to 0.9.

first_day_setup()[source]

Build the weighted graph, compute PageRank, and seed the central node.

Constructs an undirected graph-tool graph with edge weights equal to prob * intensity * layer_weight, runs PageRank, and changes the state of the node at self.quantile of the centrality distribution to STATES.I.

policy

Base policy class for the MAIS epidemic simulation framework.

This module defines the abstract Policy base class that all concrete intervention policies should subclass.

class policy.Policy(graph, model)[source]

Bases: object

Base Policy class.

To implement a custom policy, derive your subclass and override first_day_setup, run, and optionally to_df.

Parameters:

  • graph – The contact network graph object used by the simulation.

  • model – The epidemic model instance that the policy acts upon.

__init__(graph, model)[source]

Initialise the policy with a graph and model.

Parameters:

  • graph – The contact network graph object.

  • model – The epidemic model instance.

first_day_setup()[source]

Perform one-time setup on the first day the policy runs.

Called automatically by run on the first invocation. Subclasses should override this method to perform any initialisation that requires the model to be running.

run()[source]

Execute one time-step of the policy.

On the first call first_day_setup is invoked before the main logic. Subclasses should call super().run() to preserve this behaviour.

to_df()[source]

Return a DataFrame with policy-related statistics.

The returned DataFrame must contain a column T with the corresponding self.model.T date values. An empty DataFrame or None is acceptable when no statistics have been collected.

Returns:

DataFrame of policy statistics, or None if not implemented.

Return type:

pandas.DataFrame or None

quarantine_coefs

school_policy

School epidemic-intervention policies for the MAIS simulation.

This module provides policies designed for school-specific contact network graphs. They manage week/weekend school-opening schedules, optional rapid antigen testing, and class-level quarantine.

Warning

These policies are intended to be run with a special school graph only. Do not use them with general population graphs (e.g. hodoninsko, lounsko, papertown).

Classes:

BasicSchoolPolicy: Manages school open/weekend toggle and optional student testing. ClosePartPolicy: Extends BasicSchoolPolicy with the ability to close individual classes. AlternatingPolicy: Alternates two groups of classes each week. AlternateFreeMonday: Alternating policy with Monday as a free day. AlternateAndMondayPCR: Alternating policy with Monday PCR testing.

class school_policy.BasicSchoolPolicy(graph, model, config_file=None, config_obj=None)[source]

Bases: Policy

Policy that manages school-day / weekend toggling and optional testing.

On weekdays the school graph layers are active; on weekends all layers are switched off. For the first 35 simulation steps all layers are also suppressed (warm-up period). Optionally performs rapid antigen testing on configurable weekdays and places positive nodes into quarantine.

Parameters:

  • graph – The school contact network graph object. Must expose num_nodes, nodes_age, layer_weights, number_of_nodes, QUARANTINE_COEFS, nodes, is_quarantined, and nodes_class.

  • model – The epidemic model instance.

  • config_file (str, optional) –

    Path to an INI-style configuration file. The [TESTING] section may contain:

    • testing – "Yes" to enable testing (default "No").

    • sensitivity – test sensitivity in [0, 1] (default 0.4).

    • days – weekday index or list of indices on which testing is performed (default (0, 2)).

  • config_obj – Unused; reserved for future use.

__init__(graph, model, config_file=None, config_obj=None)[source]

Initialise the basic school policy.

Parameters:

  • graph – The school contact network graph object.

  • model – The epidemic model instance.

  • config_file (str, optional) – Path to a configuration file.

  • config_obj – Reserved for future use.

nodes_to_quarantine(nodes)[source]

Remove nodes from school by switching off their school-layer edges.

Sets at_school[nodes] to False and turns off all edges on school layers for those nodes.

Parameters:

nodes (numpy.ndarray) – Indices of nodes to quarantine.

nodes_from_quarantine(nodes)[source]

Return nodes to school by switching on their school-layer edges.

Sets at_school[nodes] to True and restores all edges on school layers for those nodes.

Parameters:

nodes (numpy.ndarray) – Indices of nodes to release from quarantine.

first_day_setup()[source]

Suppress all layers for the warm-up period and initialise statistics.

Saves a copy of the initial layer weights, sets all layer weights to zero (school closed during warm-up), and fills stat_in_quara with zeros for days before the policy starts.

do_testing()[source]

Perform antigen testing on configured weekdays and quarantine positives.

On test days, identifies students currently at school and stochastically classifies them as positive (with probability test_sensitivity). Positive nodes are quarantined for 7 days. Released nodes whose quarantine has ended are restored to school. Updates stat_in_quara with the current quarantine count.

stop()[source]

Signal the policy to stop any new interventions.

After calling stop, no new quarantines or school closures are initiated.

closing_and_opening()[source]

Hook for subclasses to implement dynamic school-group opening/closing.

Called each time-step after the weekend toggle. The default implementation is a no-op.

run()[source]

Execute one time-step of the school policy.

Handles first-day setup, weekend toggling, warm-up layer restoration (at day 35), subclass closing_and_opening hook, and optional testing (from day 35 onwards).

to_df()[source]

Return a DataFrame with daily school-quarantine statistics.

Returns:

DataFrame indexed by time T with column school_policy_in_quara (number of nodes in school quarantine) and day.

Return type:

pandas.DataFrame

class school_policy.ClosePartPolicy(graph, model, config_file=None, config_obj=None)[source]

Bases: BasicSchoolPolicy

School policy that can close specific classes listed in a config file.

Extends BasicSchoolPolicy with helper methods to quarantine or release all nodes belonging to a named set of classes. On first-day setup the classes listed under [CLOSED] in the config file are sent to quarantine, and optionally all teacher edges are closed.

Parameters:

  • graph – The school contact network graph object.

  • model – The epidemic model instance.

  • config_file (str, optional) –

    Path to a configuration file. The [CLOSED] section may contain:

    • close_teachers – "Yes" to close teacher edges (default "No").

    • classes – list of class names to quarantine at start.

  • config_obj – Reserved for future use.

convert_class(a)[source]

Convert node class indices to class-name strings.

Parameters:

a (numpy.ndarray) – Array of integer class indices.

Returns:

Array of class-name strings (or None for out-of-range indices).

Return type:

numpy.ndarray

nodes_in_classes(list_of_classes)[source]

Return node indices belonging to any of the specified classes.

Parameters:

list_of_classes (list[str]) – Class names to look up.

Returns:

Indices of all nodes whose class name is in list_of_classes.

Return type:

numpy.ndarray

classes_to_quarantine(list_of_classes)[source]

Send all nodes in the specified classes to quarantine.

Marks nodes as not at school and switches off their school-layer edges.

Parameters:

list_of_classes (list[str]) – Class names whose members should be quarantined.

classes_from_quarantine(list_of_classes)[source]

Release all nodes in the specified classes from quarantine.

Marks nodes as at school and switches on their school-layer edges.

Parameters:

list_of_classes (list[str]) – Class names whose members should be released.

first_day_setup()[source]

Run parent first-day setup and apply initial class closures from config.

Optionally closes teacher edges and quarantines classes listed under [CLOSED] in the configuration file.

class school_policy.AlternatingPolicy(graph, model, config_file=None)[source]

Bases: ClosePartPolicy

School policy that alternates two groups of classes week by week.

One group attends school while the other stays home; the groups swap every week (at weekend_end). Optionally, testing sub-groups can be defined to alternate which half of each group is tested on a given day.

The groups are either defined explicitly in the config file ([ALTERNATE] section with group1 and group2 class lists) or derived from the graph’s nodes_class_group attribute when use_class_groups = Yes.

Parameters:

  • graph – The school contact network graph object.

  • model – The epidemic model instance.

  • config_file (str, optional) – Path to a configuration file with [ALTERNATE] and optionally [TESTING_GROUPS] sections.

__init__(graph, model, config_file=None)[source]

Initialise the alternating policy with group definitions from config.

Parameters:

  • graph – The school contact network graph object.

  • model – The epidemic model instance.

  • config_file (str, optional) – Path to a configuration file.

closing_and_opening()[source]

Alternate active and passive groups at the start of each school week.

At weekend_end the active and passive groups are swapped: the previously active group is quarantined and the previously passive group is released. Every two weeks the testing sub-groups are also rotated.

class school_policy.AlternateFreeMonday(graph, model, config_file=None)[source]

Bases: AlternatingPolicy

Alternating policy where Monday is a free day (school starts Tuesday).

Groups alternate weekly. Testing is disabled by default.

Parameters:

  • graph – The school contact network graph object.

  • model – The epidemic model instance.

  • config_file (str, optional) – Path to a configuration file.

__init__(graph, model, config_file=None)[source]

Initialise with Monday as the first school day and no testing.

Parameters:

  • graph – The school contact network graph object.

  • model – The epidemic model instance.

  • config_file (str, optional) – Path to a configuration file.

class school_policy.AlternateAndMondayPCR(graph, model, config_file=None)[source]

Bases: AlternatingPolicy

Alternating policy with high-sensitivity (PCR-equivalent) Monday testing.

Groups alternate weekly. Testing is enabled on Mondays (weekday index 1) with a sensitivity of 0.8 (mimicking PCR).

Parameters:

  • graph – The school contact network graph object.

  • model – The epidemic model instance.

  • config_file (str, optional) – Path to a configuration file.

__init__(graph, model, config_file=None)[source]

Initialise with Monday testing at 80 % sensitivity.

Parameters:

  • graph – The school contact network graph object.

  • model – The epidemic model instance.

  • config_file (str, optional) – Path to a configuration file.

testing_policy

vaccination

Vaccination policies for the MAIS epidemic simulation.

This module provides several vaccination policy classes that vaccinate elderly and worker sub-populations according to a configurable calendar. Different subclasses implement distinct mechanisms by which vaccination reduces infection risk:

  • Vaccination – on first exposure (entering state E) a vaccinated node may be redirected back to susceptible.

  • VaccinationToR – vaccinated nodes in S are moved directly to R (recovered/immune).

  • VaccinationToA – vaccination increases the asymptomatic rate.

  • VaccinationToSA – combines the ToS and ToA mechanisms.

class vaccination.Vaccination(graph, model, config_file=None)[source]

Bases: Policy

Vaccination policy that reduces susceptibility upon exposure.

When a vaccinated node first enters state E (exposed), there is a probability (dependent on days since vaccination and whether a first or second dose has been given) that the node is returned to the susceptible state instead of progressing towards illness.

Vaccination is administered daily according to separate calendars for elderly and worker sub-populations. The number of days since vaccination is tracked per node.

Parameters:

  • graph – The contact network graph object. Must expose num_nodes, nodes_age, nodes_ecactivity, and cat_table.

  • model – The epidemic model instance.

  • config_file (str) –

    Path to an INI-style configuration file. Required. The file must include:

    • [CALENDAR] section with calendar_filename (path to a vaccination calendar CSV) and optionally delay (days between first and second dose).

    • [EFFECT] section with first_shot and second_shot effectiveness coefficients.

Raises:

str – If config_file is None or the calendar filename is missing (raises a string literal – legacy behaviour).

__init__(graph, model, config_file=None)[source]

Initialise the vaccination policy from a configuration file.

Parameters:

  • graph – The contact network graph object.

  • model – The epidemic model instance.

  • config_file (str) – Path to the required configuration file.

first_day_setup()[source]

Perform first-day setup (no-op for this policy).

stop()[source]

Signal the policy to stop vaccinating new nodes.

After calling stop, nodes already being tracked continue to have their vaccination days incremented, but no new vaccinations are administered.

move_to_S()[source]

Redirect newly exposed vaccinated nodes back to the susceptible state.

For each vaccinated node that has just entered state E for the first time today, a random draw determines whether the vaccine prevents progression. The probability depends on days-since-vaccination:

  • 14 to (delay + 6) days: first_shot_coef.

  • (delay + 7) days or more: second_shot_coef.

Nodes redirected to susceptible have their days_in_E counter reset. The count of redirected nodes is recorded in stat_moved_to_R.

process_vaccinated()[source]

Apply the vaccination effect to currently vaccinated nodes.

Calls move_to_S() to handle newly exposed vaccinated nodes. Subclasses override this method to implement alternative vaccination mechanisms.

run()[source]

Execute one time-step of the vaccination policy.

Increments vaccination-day counters, applies the vaccination effect, and administers new vaccinations according to the daily calendar.

vaccinate_old(num)[source]

Vaccinate up to num elderly nodes (sorted by descending age).

Nodes that are already vaccinated, currently detected as active cases, or dead are skipped.

Parameters:

num (int) – Maximum number of elderly nodes to vaccinate today.

vaccinate_workers(num)[source]

Vaccinate up to num worker nodes chosen at random.

Nodes that are currently detected as active cases or dead are excluded from selection. If fewer eligible workers than num exist, all eligible workers are vaccinated.

Parameters:

num (int) – Target number of workers to vaccinate today.

to_df()[source]

Return a DataFrame with daily vaccination statistics.

Returns:

DataFrame indexed by time T with column moved_to_R (nodes redirected to susceptible/ recovered each day) and day.

Return type:

pandas.DataFrame

class vaccination.VaccinationToR(graph, model, config_file=None)[source]

Bases: Vaccination

Vaccination policy that moves susceptible nodes directly to recovered.

On day 14 after the first shot, susceptible vaccinated nodes are moved to R with probability first_shot_coef. On day delay + 7 after the first shot, a further fraction (second_shot_coef - first_shot_coef) is moved to R.

Parameters:

  • graph – The contact network graph object.

  • model – The epidemic model instance.

  • config_file (str) – Path to the required configuration file.

process_vaccinated()[source]

Move eligible susceptible vaccinated nodes to recovered state.

Applies first-shot and second-shot effects by drawing random numbers and calling model.move_target_nodes_to_R.

class vaccination.VaccinationToA(graph, model, config_file=None)[source]

Bases: Vaccination

Vaccination policy that increases the asymptomatic rate of vaccinated nodes.

Fourteen days after the first shot the asymptomatic rate is updated to reflect first-dose effectiveness. A further update occurs at delay + 7 days to reflect second-dose effectiveness.

Parameters:

  • graph – The contact network graph object.

  • model – The epidemic model instance.

  • config_file (str) – Path to the required configuration file.

update_asymptomatic_rates()[source]

Update model.asymptomatic_rate for nodes at key vaccination milestones.

First-shot effect is applied at day 14; second-shot effect at day self.delay + 7.

process_vaccinated()[source]

Apply vaccination effect by updating asymptomatic rates.

class vaccination.VaccinationToSA(graph, model, config_file=None)[source]

Bases: VaccinationToA

Vaccination policy combining susceptible redirection and asymptomatic-rate update.

Applies both the Vaccination.move_to_S() mechanism (redirecting newly exposed vaccinated nodes back to susceptible) and the VaccinationToA.update_asymptomatic_rates() mechanism each time-step.

Parameters:

  • graph – The contact network graph object.

  • model – The epidemic model instance.

  • config_file (str) – Path to the required configuration file.

process_vaccinated()[source]

Apply both susceptible-redirection and asymptomatic-rate-update effects.

wee_cold

wee_cold_sim

src/utils

config_utils

Configuration file utilities for the MAIS simulation.

This module provides classes and helpers for reading, writing, and generating INI-style configuration files used to parameterise simulation runs.

Key components:

  • string_to_value(): Type-coercing string parser used when reading INI values.

  • ConfigFile: Thin wrapper around configparser.ConfigParser for loading, saving, and querying individual INI files.

  • ConfigFileGenerator: Expands a template INI file containing semicolon-separated parameter lists into a stream of fully-specified ConfigFile instances, one per parameter combination.

config_utils.string_to_value(s)[source]

Convert a raw INI string value to the most appropriate Python type.

Tries type conversions in the following order:

  1. int – if the entire string represents an integer.

  2. float – if the entire string represents a floating-point number.

  3. list – if the string contains a comma; it is split on commas and each token is stripped of surrounding whitespace.

  4. str – the original string is returned unchanged.

Parameters:

s (str) – Raw string value read from a configuration file.

Returns:

The converted value.

Return type:

int or float or list of str or str

class config_utils.ConfigFile(param_dict=None)[source]

Bases: object

Wrapper around configparser.ConfigParser for INI-style config files.

Provides convenience methods for loading from and saving to .ini files, serialising to a string, and reading sections as type-converted dictionaries. Key-case is preserved (optionxform = str).

Parameters:

param_dict (dict, optional) – If provided, a mapping of {section_name: {key: value}} pairs used to pre-populate the underlying configparser.ConfigParser. Defaults to None (empty configuration).

save(filename)[source]

Write the configuration to a file or file-like object.

Parameters:

filename (str or file-like) – If a string, the configuration is written to that file path (UTF-8 encoded). Otherwise the object is treated as a writable file-like and config.write() is called on it directly.

to_string()[source]

Serialise the configuration to an INI-formatted string.

Returns:

The full INI representation of the configuration, equivalent to what would be written by save().

Return type:

str

load(filename)[source]

Read a configuration from an INI file.

Parameters:

filename (str) – Path to the .ini file to read.

Raises:

ValueError – If filename does not exist on the filesystem.

section_as_dict(section_name)[source]

Return the contents of a section as a type-converted dictionary.

Each raw string value in the section is converted by string_to_value() to the most appropriate Python type.

Parameters:

section_name (str) – Name of the INI section to retrieve.

Returns:

Mapping of {key (str): value} for all entries in the section, with values converted by string_to_value(). Returns an empty dict if the section does not exist.

Return type:

dict

fix_output_id()[source]

Resolve and replace the OUTPUT_ID.id field with a descriptive string.

Reads the id entry from the [OUTPUT_ID] section. If it contains one or more section:key references (as a list or a single string), each reference is resolved to its current value in the configuration, and a composite identifier string of the form _Section_key=value is constructed and stored back into OUTPUT_ID.id. Spaces in the resulting string are replaced with underscores.

If OUTPUT_ID.id is not present, the method returns without making any changes.

class config_utils.ConfigFileGenerator[source]

Bases: object

Generator that expands a template INI file into individual ConfigFile instances.

The template INI file may contain semicolon-separated lists of values for any key. The generator computes the Cartesian product of all such lists (using sklearn.model_selection.ParameterGrid) and yields one fully-specified ConfigFile per parameter combination.

After each config file is generated, ConfigFile.fix_output_id() is called to resolve any OUTPUT_ID references.

load(filename)[source]

Load a template INI file and yield one ConfigFile per parameter combination.

Each key that contains a ';'-separated list of values is treated as a parameter with multiple options. The full Cartesian product of all such options (across all keys and sections) is enumerated, and each combination is yielded as a separate ConfigFile.

Parameters:

filename (str) – Path to the template .ini file. The file may contain semicolon-separated lists of values for any key.

Yields:

ConfigFile – A fully-specified configuration for one parameter combination, with OUTPUT_ID resolved via ConfigFile.fix_output_id().

Raises:

ValueError – If filename does not exist on the filesystem.

global_configs

Global configuration parameters and utilities for the MAIS simulation.

This module provides module-level constants that act as global toggles and settings used throughout the simulation. It also exposes a lightweight monitoring helper that emits structured log messages for a designated node.

These globals should only be mutated when truly necessary; prefer passing configuration explicitly through function arguments or config objects.

global_configs.monitor(t, msg)[source]

Emit a monitoring log message for the globally watched node.

Logs an INFO-level message that identifies the current simulation day, the monitored node (MONITOR_NODE), and an arbitrary status message. The function is a no-op in terms of return value; its purpose is purely for diagnostic logging during a simulation run.

Parameters:

  • t (int) – The current simulation day (time-step index).

  • msg (str) – A descriptive message about the node’s current status or event to be recorded.

graph_utils

Graph analysis utilities for the MAIS simulation.

This module provides helper functions for computing structural properties of the contact graph used in the simulation, such as node degree statistics derived from edge-probability data.

graph_utils.compute_mean_degree(graph, nodes)[source]

Compute the mean expected number of contacts (degree) for a set of nodes.

For each pair of nodes in the graph, the function aggregates all edges connecting them to compute the probability that at least one contact occurs on any layer. This probability is stored in a dense sparse matrix, and the expected degree of each node in nodes is the row sum of that matrix. The mean is then taken over all nodes in nodes.

The computation iterates over all node pairs, so it is suited for analysis rather than performance-critical simulation paths.

Parameters:

  • graph (LightGraph) – The contact graph object. Must expose: - graph.num_nodes (int): Total number of nodes. - graph.nodes (iterable): All node indices. - graph.A (array-like): Adjacency structure mapping a node pair (n1, n2) to an index into graph.edges_repo (0 means no edge). - graph.edges_repo (list): Repository of edge collections keyed by the index returned from graph.A. - graph.get_edges_probs(edges) (callable): Returns an array of transmission probabilities for the given edges.

  • nodes (iterable) – Subset of node indices for which the mean degree is computed.

Returns:

Mean expected number of contacts per node over the given nodes.

Return type:

float

history_utils

Time-series and history container classes for the MAIS simulation.

This module provides a hierarchy of array-backed containers that record simulation state over time. Containers grow automatically when new entries are appended beyond their initial capacity, and they can be trimmed to the actual simulation length at the end of a run.

class history_utils.BaseSeries[source]

Bases: object

Base class for all time-series containers in the MAIS simulation.

A BaseSeries stores values produced sequentially over simulation time—one item per iteration. Concrete subclasses back the storage with a NumPy array and override bloat() to extend that array when needed.

The container supports standard index-based read/write access. On a write to the index immediately past the current end of the array, the array is automatically extended by 100 elements before the write proceeds.

__getitem__(idx)[source]

Return the value at idx.

Parameters:

idx (int or slice) – Index or slice into the underlying array.

Returns:

The element (or sub-array) at the given position.

__setitem__(idx, data)[source]

Set the value at idx, auto-extending the array if necessary.

If idx equals the current length of the underlying array, the array is grown by 100 elements via bloat() before the assignment is made.

Parameters:

  • idx (int) – Target index. Must be <= len(self.values).

  • data – Value to store at position idx.

Raises:

IndexError – If idx is out of range and not equal to len(self.values).

save(filename)[source]

Save the underlying array to a NumPy .npy file.

Parameters:

filename (str) – Destination file path (passed to numpy.save).

__len__()[source]

Return the number of elements in the series.

Returns:

Length of the underlying array.

Return type:

int

len()[source]

Return the number of elements in the series.

Returns:

Length of the underlying array.

Return type:

int

asarray()[source]

Return the underlying NumPy array directly.

Returns:

The internal storage array.

Return type:

numpy.ndarray

bloat(len)[source]

Extend the internal storage by len elements.

Subclasses must override this method with an implementation that appends len default-initialised elements to self.values.

Parameters:

len (int) – Number of elements to add.

class history_utils.TimeSeries(len, dtype=<class 'float'>)[source]

Bases: BaseSeries

One-dimensional time series backed by a NumPy array of a given dtype.

Stores one scalar value per simulation time-step. The array is pre-allocated with zeros and is extended automatically if needed.

Parameters:

  • len (int) – Initial capacity (number of time-steps to pre-allocate).

  • dtype (type, optional) – NumPy dtype for the underlying array. Defaults to float.

bloat(len)[source]

Extend the series by len zero-initialised elements.

Parameters:

len (int) – Number of additional zero elements to append.

finalize(tidx)[source]

Trim the series to the actual simulation length.

Removes trailing zero-padding so that the array ends at the last recorded time-step.

Parameters:

tidx (int) – Zero-based index of the last recorded time-step. Elements beyond this index are discarded.

get_values()[source]

Return the underlying NumPy array.

Returns:

The internal 1-D storage array.

Return type:

numpy.ndarray

class history_utils.TransitionHistory(len, dtype=<class 'int'>, width=3)[source]

Bases: BaseSeries

Two-dimensional history table recording state-transition events over time.

Each row corresponds to one recorded event (or time-step), and the fixed number of columns (width) encodes the details of each transition (e.g., node ID, source state, target state). The row count grows automatically; the column count is fixed at construction.

Parameters:

  • len (int) – Initial row capacity (number of events to pre-allocate).

  • dtype (type, optional) – NumPy dtype for the underlying 2-D array. Defaults to int.

  • width (int, optional) – Number of columns per row. Defaults to 3.

bloat(len)[source]

Extend the table by len zero-initialised rows.

Parameters:

len (int) – Number of additional rows to append.

finalize(tidx)[source]

Trim the table to the actual number of recorded events.

Removes trailing zero-padded rows so that the table contains only the rows up to and including tidx.

Parameters:

tidx (int) – Zero-based index of the last recorded event. Rows beyond this index are discarded.

class history_utils.ShortListSeries(length)[source]

Bases: object

Fixed-capacity FIFO list that discards the oldest element when full.

Behaves like a sliding window: once the list reaches length items, appending a new value automatically removes the oldest one (index 0). Useful for maintaining a rolling window of recent simulation metrics.

Parameters:

length (int) – Maximum number of elements to retain.

append(member)[source]

Append member and evict the oldest element if at capacity.

Parameters:

member – The value to add to the end of the list.

__getitem__(idx)[source]

Return the element at idx.

Parameters:

idx (int or slice) – Index into the internal list.

Returns:

The element (or sub-list) at the given position.

__len__()[source]

Return the current number of stored elements.

Returns:

Number of elements currently in the list.

Return type:

int

plot_utils

Plotting utilities for MAIS simulation histories.

This module provides functions for loading simulation output CSV files and visualising epidemic curves and other time-series metrics via Matplotlib and Seaborn. It supports single-run plots, multi-run aggregated line plots, and animated state-histogram views.

Public API:
plot_utils.plot_history(filename: str)[source]

Plot the all_infectious curve from a single simulation history CSV.

Loads the history file, plots all_infectious against T using the default Pandas/Matplotlib backend, and displays the figure interactively.

Parameters:

filename (str) – Path to the simulation output CSV file.

plot_utils.plot_histories(*args, group_days: int = None, group_func: str = 'max', **kwargs)[source]

Plot all_infectious from multiple simulation history CSV files.

Loads each history file, optionally groups records into day-buckets, then overlays all runs on a single line plot using _plot_lineplot().

Parameters:

  • *args (str) – One or more paths to simulation output CSV files.

  • group_days (int, optional) – If set, aggregates rows into buckets of this many days before plotting. Defaults to None (no grouping).

  • group_func (str, optional) – Aggregation function applied within each day bucket (e.g., "max", "mean"). Defaults to "max".

  • **kwargs – Additional keyword arguments forwarded to _plot_lineplot() (e.g., title, save_path).

plot_utils.plot_mutliple_policies(policy_dict: Dict[str, List[str]], group_days: int = None, group_func: str = 'max', value='all_infectious', max_days=None, **kwargs)[source]

Compare a single metric across multiple policies on one line plot.

For each policy, loads all associated history files, concatenates them, and renders a median line with inter-quartile shading using _plot_lineplot().

Parameters:

  • policy_dict (Dict[str, List[str]]) – Mapping of policy name to a list of history CSV file paths for that policy.

  • group_days (int, optional) – Day-bucket size for temporal aggregation. Defaults to None (no grouping).

  • group_func (str, optional) – Aggregation function applied per bucket. Defaults to "max".

  • value (str, optional) – Column name of the metric to plot on the y-axis. Defaults to "all_infectious".

  • max_days (int, optional) – If set, truncates each history to the first max_days rows. Defaults to None.

  • **kwargs – Additional keyword arguments forwarded to _plot_lineplot().

plot_utils.plot_mutliple_policies_everything(policy_dict: Dict[str, List[str]], group_days: int = None, group_func: str = 'max', max_days=None, **kwargs)[source]

Render a multi-panel comparison of all tracked metrics across multiple policies.

Loads and concatenates histories for every policy, then delegates to either _plot_lineplot2() (variant 2) or _plot_lineplot3() (default) depending on the optional variant keyword argument.

Parameters:

  • policy_dict (Dict[str, List[str]]) – Mapping of policy name to a list of history CSV file paths for that policy.

  • group_days (int, optional) – Day-bucket size for temporal aggregation. Defaults to None (no grouping).

  • group_func (str, optional) – Aggregation function applied per bucket. Defaults to "max".

  • max_days (int, optional) – If set, truncates each history to the first max_days rows. Defaults to None.

  • **kwargs – Additional keyword arguments forwarded to the chosen plot function. The special key variant (int) selects the plot layout (2 selects _plot_lineplot2(); any other value selects _plot_lineplot3()). The title key is required by the underlying plot functions.

plot_utils.plot_state_histogram(filename: str, title: str = 'Simulation', states: List[str] = None, save_path: str = None)[source]

Render an animated bar chart showing the per-state population over time.

Reads the given history CSV and produces an animation where each frame corresponds to one simulation day. Each bar represents a disease/model state, and its height equals the number of nodes in that state on the corresponding day.

Parameters:

  • filename (str) – Path to the simulation output CSV file.

  • title (str, optional) – Base title displayed in the figure. The current day number is appended dynamically per frame. Defaults to "Simulation".

  • states (List[str], optional) – Subset of state column names to include in the histogram. If None, all state columns present in the CSV (excluding metadata columns) are shown. Defaults to None.

  • save_path (str, optional) – If provided, the animation is saved to this file path using FFMpeg at 10 fps before being displayed. Defaults to None.

policy_utils

Policy and scenario loading utilities for the MAIS simulation.

This module provides functions for loading simulation scenario definitions from CSV files and converting them to dictionaries that can drive per-day policy changes (such as contact-layer weight adjustments) during a run.

policy_utils.load_scenario_dict(filename: str, sep=',', return_data='list')[source]

Load a scenario definition from a CSV file and return it as a dictionary.

The CSV file is expected to have at minimum an id column, a name column, and one or more numeric day-columns. Each day-column represents a particular simulation day, and the rows represent the layers (or other entities) whose weights/values are being specified for that day.

Parameters:

  • filename (str) – Path to the CSV file to read.

  • sep (str, optional) – Column delimiter used in the CSV file. Defaults to ','.

  • return_data (str, optional) –

    Format for the per-day values:

    • 'list': Each day maps to a plain Python list of values ordered by row.

    • 'names': Each day maps to a {name: value} dictionary keyed by the name column.

    • 'ids': Each day maps to a {name: value} dictionary keyed by the name column (same as 'names' in current implementation).

    Defaults to 'list'.

Returns:

Mapping of {day (int): output} where output is formatted according to return_data.

Return type:

dict

Raises:

ValueError – If return_data is not one of 'list', 'names', or 'ids'.

pool

Multiprocessing worker pool for parallel model evaluation in the MAIS simulation.

This module provides a simple process-pool abstraction built on top of multiprocessing. Each worker process runs an evaluation function in an infinite loop, consuming queries from a shared input queue and posting results to a shared output queue. The main process enqueues tasks and dequeues results through the Pool interface.

pool.worker(name, evalfunc, querries, answers, model)[source]

Entry point for a pool worker process.

Runs an infinite loop: retrieves one query at a time from the querries queue, evaluates evalfunc(model, query), and places the result onto the answers queue. The loop continues until the process is terminated externally (e.g., via Pool.close()).

Parameters:

  • name (int) – Numeric identifier for this worker (used for logging or debugging).

  • evalfunc (callable) – Function to evaluate. Called as evalfunc(model, query) and must return a serialisable result.

  • querries (multiprocessing.Queue) – Input queue from which queries are consumed.

  • answers (multiprocessing.Queue) – Output queue to which computed answers are posted.

  • model – Model object passed as the first argument to evalfunc. Each worker receives its own model instance.

class pool.Pool(processors, evalfunc, models)[source]

Bases: object

Process pool that parallelises model evaluation across multiple workers.

Workers consume tasks from a shared queries queue and post results to a shared answers queue. The pool is non-blocking from the caller’s perspective: tasks are submitted via putQuerry() and results are retrieved (blocking) via getAnswer().

Parameters:

  • processors (int) – Number of worker processes to spawn.

  • evalfunc (callable) – Evaluation function passed to each worker. Signature: evalfunc(model, query) -> answer.

  • models (list) – List of model instances, one per worker. Element i is passed to worker i.

putQuerry(querry)[source]

Enqueue a query for processing by one of the worker processes.

Parameters:

querry – The query object to evaluate. Must be picklable.

getAnswer()[source]

Block until an answer is available and return it.

Returns:

The result produced by evalfunc for the oldest unread query.

close()[source]

Terminate all worker processes and shut down the pool.

Sends a SIGTERM signal to each worker process. After calling this method the pool should not be used further.

random_utils

Random number generation utilities for the MAIS simulation.

This module provides helpers for reproducible random sampling, ordered-tuple generation, and discrete duration sampling. It is used across the simulation to draw stochastic values for disease progression durations and other time-varying random quantities.

class random_utils.RandomGenerator(seed)[source]

Bases: object

Seeded pseudo-random number generator wrapper using the SFC64 bit generator.

Wraps NumPy’s Generator backed by the fast SFC64 bit generator so that independent, reproducible streams can be attached to different parts of the model.

Note

Adding per-component generators is still in progress.

Parameters:

seed (int) – Integer seed passed to SFC64 for reproducibility.

rand()[source]

Generate n uniformly distributed random floats in [0, 1).

Parameters:

n (int) – Number of random values to generate.

Returns:

Array of shape (n,) with values in [0, 1).

Return type:

numpy.ndarray

random_utils.gen_tuple1(n, shape, *args)[source]

Generate an n-tuple of random values satisfying a strict ordering.

Draws values (r_1, r_2, ..., r_n) such that r_1 < r_2 < ... < r_n element-wise across a batch of size shape. Any positions that violate the ordering are resampled repeatedly until all positions satisfy the constraint.

Parameters:

  • n (int) – Number of elements in the tuple. Must equal len(args).

  • shape (int or tuple) – Shape of the batch to generate (passed as n argument to each generator’s get method).

  • *args – Exactly n random duration generator objects, each exposing a get(n=...) method that returns a NumPy array of samples.

Returns:

List of n arrays, each of shape shape, satisfying result[0] < result[1] < ... < result[n-1] element-wise.

Return type:

list of numpy.ndarray

Example

>>> gen_tuple(3, rng1, rng2, rng3)
random_utils.gen_tuple2(n, shape, *args)[source]

Generate an n-tuple of random values satisfying a strict ordering (clipping variant).

Draws values (r_1, r_2, ..., r_n) such that r_1 < r_2 < ... < r_n element-wise across a batch of size shape. Unlike gen_tuple1(), ordering is enforced by clipping each subsequent value to be at least previous + 1 rather than by resampling.

Parameters:

  • n (int) – Number of elements in the tuple. Must equal len(args).

  • shape (int or tuple) – Shape of the batch to generate (passed as n argument to each generator’s get method).

  • *args – Exactly n random duration generator objects, each exposing a get(n=...) method that returns a NumPy array of samples.

Returns:

List of n arrays, each of shape shape, satisfying result[0] < result[1] < ... < result[n-1] element-wise.

Return type:

list of numpy.ndarray

Example

>>> gen_tuple(3, rng1, rng2, rng3)
random_utils.gen_tuple(n, shape, *args)[source]

Generate an n-tuple of strictly ordered random values.

Delegates to gen_tuple2(). See that function for full documentation.

Parameters:

  • n (int) – Number of elements in the tuple.

  • shape (int or tuple) – Shape of the batch to generate.

  • *args – Exactly n random duration generator objects.

Returns:

List of n strictly ordered arrays.

Return type:

list of numpy.ndarray

class random_utils.RandomDuration(probs, precompute=False)[source]

Bases: object

Discrete random duration sampler driven by a full probability distribution.

Intended for generating the time (in discrete steps, e.g. days) that an agent spends in a particular disease state. The distribution is specified as a probability mass function (PMF) over non-negative integer durations starting from zero.

Parameters:

  • probs (array-like) – NumPy array of probabilities for durations 0, 1, 2, ..., len(probs)-1. Values must be non-negative and sum to 1.

  • precompute (bool, optional) – If True, a large buffer of 10^6 pre-drawn samples is generated at construction time. Currently the buffer is not stored, so this flag has no effect on subsequent get calls. Defaults to False.

get(n=1)[source]

Draw n random duration values from the distribution.

Parameters:

n (int, optional) – Number of samples to draw. Defaults to 1.

Returns:

Array of n integer duration values drawn according to self.probs.

Return type:

numpy.ndarray

sparse_utils

Sparse-matrix utility functions for the MAIS simulation.

This module is no longer used in current versions of the model. It was originally used when a contact graph was compressed into a sparse matrix where multi-edges were aggregated into single weighted entries.

The functions operate on SciPy CSR/CSC sparse matrices and provide element-wise row/column scaling, row/column product aggregation, and a specialised element-wise multiplication that treats structurally-zero entries as ones (useful for probability-complement arithmetic).

sparse_utils.multiply_row(A, row_idx, alpha, trunc=False)[source]

Scale all stored values in a single row of a CSR matrix in place.

Only the explicitly stored (non-zero) entries in the row are affected. Structural zeros are untouched.

Parameters:

  • A (scipy.sparse.csr_matrix) – The sparse matrix to modify in place. Must be in CSR format.

  • row_idx (int) – Zero-based index of the row to scale.

  • alpha (float) – Scalar multiplier applied to every stored entry in the specified row.

  • trunc (bool, optional) – If True, the scaled values are clipped to the interval [0.0, 1.0] after multiplication. Defaults to False.

sparse_utils.multiply_col(A, col_idx, alpha, trunc=False)[source]

Scale all stored values in a single column of a CSR matrix in place.

Locates every explicitly stored entry in the given column and multiplies it by alpha. Works on CSR format by scanning the indices array.

Parameters:

  • A (scipy.sparse.csr_matrix) – The sparse matrix to modify in place. Must be in CSR format.

  • col_idx (int) – Zero-based index of the column to scale.

  • alpha (float) – Scalar multiplier applied to every stored entry in the specified column.

  • trunc (bool, optional) – If True, the scaled values are clipped to the interval [0.0, 1.0] after multiplication. Defaults to False.

sparse_utils.prop_of_row(A)[source]

Compute the product of stored values in each row of a CSR matrix.

For each row the function multiplies all explicitly stored (non-zero) entries together. Rows with no stored entries retain a product of 1.0 (identity for multiplication), which corresponds to the convention that missing entries represent the value 1.

Parameters:

A (scipy.sparse.csr_matrix) – Input sparse matrix in CSR format.

Returns:

A 1-D array of shape (A.shape[0],) where element i is the product of all stored values in row i.

Return type:

numpy.ndarray

sparse_utils.prop_of_column(A)[source]

Compute the product of stored values in each column of a CSR matrix.

For each unique column index present in the matrix the function multiplies all explicitly stored entries in that column. Columns with no stored entries retain a product of 1.0.

Parameters:

A (scipy.sparse.csr_matrix) – Input sparse matrix in CSR format.

Returns:

A 1-D array of shape (A.shape[1],) where element j is the product of all stored values in column j.

Return type:

numpy.ndarray

sparse_utils.multiply_zeros_as_ones(a, b)[source]

Element-wise multiply two sparse matrices treating structural zeros as ones.

Standard sparse element-wise multiplication treats structurally-zero positions as 0 * 0 = 0. This function instead treats a missing entry (structural zero) in either matrix as the value 1.0, so that:

  • positions present in both a and b → a[i,j] * b[i,j]

  • positions present only in a → a[i,j] (b treated as 1)

  • positions present only in b → b[i,j] (a treated as 1)

  • positions absent in both → 0 (stored as structural zero)

This is useful for computing the joint probability of no contact across multiple probability layers, where an absent entry means “no edge, hence probability 1 of no contact on this layer”.

Parameters:

  • a (scipy.sparse.csr_matrix) – First sparse matrix.

  • b (scipy.sparse.csr_matrix) – Second sparse matrix. Must have the same shape as a.

Returns:

Result matrix with the same shape as a and b, where element-wise multiplication respects the zeros-as-ones convention described above.

Return type:

scipy.sparse.csr_matrix