Performing simulations#
How does the simulator work?#
The simulator is a class that allows previously designed experiments to be executed. To run it, we need a network (multilayer or temporal; it may have single layer as well) and a corresponding model. After the experiment is completed, the user can view the results as a report and a visualisation of the actor states.
The following example considers a custom SIR~UA model: the spread of two processes in separate layers that are dependent on each other. * “contagion” process with states S (suspected), I (infected), R (removed), * “awareness” process with states U (unaware), A (aware). The possible transitions can be described by the following graph:
S·U---->I·U---->R·U
| | |
| | |
v v v
S·A---->I·A---->R·A
All transitions except I->R are determined by interactions between neighbouring nodes. Nodes can transition from I to R without any external stimuli. The constructor parameters are probabilities of transitions and the initial percentages of infected and aware nodes.
Example of usage#
Import necessary libraries
import networkx as nx
import network_diffusion as nd
import numpy as np
from network_diffusion.models import (
BaseModel,
CompartmentalGraph,
NetworkUpdateBuffer,
)
Create the propagation model
1class SIR_UAModel(BaseModel):
2
3 def __init__(
4 self,
5 alpha: float,
6 alpha_prime: float,
7 beta: float,
8 gamma: float,
9 delta: float,
10 ill_seeds: float,
11 aware_seeds: float,
12 ) -> None:
13 """
14 An SIR~UA model.
15
16 :param alpha: probability weight of S->I for unaware nodes
17 :param alpha_prime: prob. of S->I for aware nodes
18 :param beta: prob. of I->R for both unaware and aware nodes
19 :param gamma: prob. of U->A for both suspected and removed nodes
20 :param delta: prob. of U->A for ill nodes
21 :param ill_seeds: % of initially I nodes
22 :param aware_seeds: % of initially A nodes
23 """
24 compartments = self._create_compartments(
25 alpha=alpha,
26 alpha_prime=alpha_prime,
27 beta=beta,
28 gamma=gamma,
29 delta=delta,
30 ill_seeds=ill_seeds,
31 aware_seeds=aware_seeds,
32 )
33 self.__comp_graph = compartments
34 self.__seed_selector = nd.seeding.RandomSeedSelector()
35
36 @staticmethod
37 def _create_compartments(
38 alpha: float,
39 alpha_prime: float,
40 beta: float,
41 gamma: float,
42 delta: float,
43 ill_seeds: float,
44 aware_seeds: float,
45 ) -> CompartmentalGraph:
46 # define processes, allowed states and initial percentages of actors in those states
47 phenomena: dict[str, tuple] = {
48 "contagion": (["S", "I", "R"], [100 - ill_seeds, ill_seeds, 0]),
49 "awareness": (["U", "A"], [100 - aware_seeds, aware_seeds]),
50 }
51
52 # wrap them into compartments
53 cg = CompartmentalGraph()
54 for phenomenon, (states, budget) in phenomena.items():
55 cg.add(process_name=phenomenon, states=states) # name of process
56 cg.seeding_budget.update({phenomenon: budget}) # initial %s
57 cg.compile(background_weight=0)
58
59 # set up weights of transitions for SIR and unaware
60 cg.set_transition_fast(
61 "contagion.S", "contagion.I", ("awareness.U",), alpha
62 )
63 cg.set_transition_fast(
64 "contagion.I", "contagion.R", ("awareness.U",), beta
65 )
66
67 # set up weights of transitions for SIR and aware
68 cg.set_transition_fast(
69 "contagion.S", "contagion.I", ("awareness.A",), alpha_prime
70 )
71 cg.set_transition_fast(
72 "contagion.I", "contagion.R", ("awareness.A",), beta
73 )
74
75 # set up weights of transitions for UA and suspected
76 cg.set_transition_fast(
77 "awareness.U", "awareness.A", ("contagion.S",), gamma
78 )
79
80 # set up weights of transitions for UA and infected
81 cg.set_transition_fast(
82 "awareness.U", "awareness.A", ("contagion.I",), delta
83 )
84
85 # set up weights of transitions for UA and removed
86 cg.set_transition_fast(
87 "awareness.U", "awareness.A", ("contagion.R",), gamma
88 )
89
90 return cg
91
92 @property
93 def _compartmental_graph(self) -> CompartmentalGraph:
94 """Compartmental model that defines allowed transitions and states."""
95 return self.__comp_graph
96
97 @property
98 def _seed_selector(self) -> nd.seeding.RandomSeedSelector:
99 """A method of selecting seed agents."""
100 return self.__seed_selector
101
102 def __str__(self) -> str:
103 descr = f"{nd.utils.BOLD_UNDERLINE}\n"
104 descr += f"SIR-UA Model\n"
105 descr += self._compartmental_graph.__str__()
106 descr += str(self._seed_selector)
107 return descr
108
109 def determine_initial_states(
110 self, net: nd.MultilayerNetwork
111 ) -> list[NetworkUpdateBuffer]:
112 if not net.is_multiplex():
113 raise ValueError("This model works only with multiplex networks!")
114
115 budget = self._compartmental_graph.get_seeding_budget_for_network(net)
116 nodes_ranking = self._seed_selector.nodewise(net)
117 initial_states = []
118
119 # set initial states in contagion process/layer
120 for node_position, node_id in enumerate(nodes_ranking["contagion"]):
121 if node_position < budget["contagion"]["I"]:
122 node_initial_state = "I"
123 else:
124 node_initial_state = "S"
125 initial_states.append(
126 nd.models.NetworkUpdateBuffer(
127 node_id, "contagion", node_initial_state
128 )
129 )
130
131 # set initial states in awareness process/layer
132 for node_position, node_id in enumerate(nodes_ranking["awareness"]):
133 if node_position < budget["awareness"]["A"]:
134 node_initial_state = "A"
135 else:
136 node_initial_state = "U"
137 initial_states.append(
138 nd.models.NetworkUpdateBuffer(
139 node_id, "awareness", node_initial_state
140 )
141 )
142
143 return initial_states
144
145 @staticmethod
146 def flip_a_coin(prob_success: float) -> bool:
147 result = np.random.choice([0, 1], p=[1 - prob_success, prob_success])
148 if result == 1:
149 return True
150 return False
151
152 def agent_evaluation_step(
153 self, agent: int, layer_name: str, net: nd.MultilayerNetwork
154 ) -> str:
155 layer_graph: nx.Graph = net[layer_name]
156
157 # Get possible transitions for the node's state
158 current_state = layer_graph.nodes[agent]["status"]
159 transitions = self._compartmental_graph.get_possible_transitions(
160 net.get_actor(agent).states_as_compartmental_graph(), layer_name
161 )
162
163 # If there is no possible transition, return current state
164 if len(transitions) == 0:
165 return current_state
166
167 # If transition does not depend on interactions with neighbours (i.e. I->R)
168 if layer_name == "contagion" and current_state == "I":
169 new_state = "R"
170 if self.flip_a_coin(transitions[new_state]):
171 return new_state
172
173 # Otherwise iterate through neighbours
174 else:
175 for neighbour in nx.neighbors(layer_graph, agent):
176 new_state = layer_graph.nodes[neighbour]["status"]
177 if new_state in transitions and self.flip_a_coin(
178 transitions[new_state]
179 ):
180 return new_state
181
182 return current_state
183
184 def network_evaluation_step(
185 self, net: nd.MultilayerNetwork
186 ) -> list[NetworkUpdateBuffer]:
187 new_states = []
188 for layer_name, layer_graph in net.layers.items():
189 for node in layer_graph.nodes():
190 new_state = self.agent_evaluation_step(node, layer_name, net)
191 layer_graph.nodes[node]["status"] = new_state
192 new_states.append(
193 nd.models.NetworkUpdateBuffer(node, layer_name, new_state)
194 )
195 return new_states
196
197 def get_allowed_states(
198 self, net: nd.MultilayerNetwork
199 ) -> dict[str, tuple[str, ...]]:
200 return self._compartmental_graph.get_compartments()
Load the network
net = nd.MultilayerNetwork.from_nx_layer(
nx.karate_club_graph(), ["contagion", "awareness"]
)
print(net)
============================================
network parameters
--------------------------------------------
general parameters:
number of layers: 2
number of actors: 34
number of nodes: 68
number of edges: 156
layer 'contagion' parameters:
graph type - <class 'networkx.classes.graph.Graph'>
number of nodes - 34
number of edges - 78
average degree - 4.5882
clustering coefficient - 0.5706
layer 'awareness' parameters:
graph type - <class 'networkx.classes.graph.Graph'>
number of nodes - 34
number of edges - 78
average degree - 4.5882
clustering coefficient - 0.5706
============================================
Initialise an instance of the propagation model
model = SIR_UAModel(
alpha=0.19,
alpha_prime=0.0665,
beta=0.1,
gamma=0.01,
delta=0.71,
ill_seeds=5,
aware_seeds=5,
)
print(model)
============================================
SIR-UA Model
============================================
compartmental model
--------------------------------------------
processes, their states and initial sizes:
'contagion': [S:95%, I:5%, R:0%]
'awareness': [U:95%, A:5%]
--------------------------------------------
process 'contagion' transitions with nonzero weight:
from S to I with probability 0.19 and constrains ['awareness.U']
from I to R with probability 0.1 and constrains ['awareness.U']
from S to I with probability 0.0665 and constrains ['awareness.A']
from I to R with probability 0.1 and constrains ['awareness.A']
--------------------------------------------
process 'awareness' transitions with nonzero weight:
from U to A with probability 0.01 and constrains ['contagion.S']
from U to A with probability 0.71 and constrains ['contagion.I']
from U to A with probability 0.01 and constrains ['contagion.R']
============================================
============================================
seed selection method
--------------------------------------------
nodewise random choice
============================================
Perform the simulation
experiment = nd.Simulator(model, net)
run_logs = experiment.perform_propagation(n_epochs=10)
Save experiment results. User is able to save them to the file or print them out
run_logs.report(path="results", visualisation=True)
The logs contain: * a description of the network (txt file), * a description of the propagation model (txt file), * a report of the spreading for all simulated phenomena (separate csv files), * a capture of states of every single node at the end of each simulation step (JSON file), * a brief visualisation of propagation.
If you need to process the results directly in Python, you can extract them with two functions. For aggregated results for each process
run_logs.get_aggragated_logs()
or for detailed logs concerning all nodes.
run_logs.get_detailed_logs()