import random
from typing import Dict, List, Union
from episimmer.policy.lockdown_policy import EventLockdown
from episimmer.read_file import (ReadAgents, ReadEvents, ReadLocations,
ReadOneTimeEvents)
from episimmer.simulate import Simulate
from episimmer.utils.time import Time
from episimmer.vulnerability_detection.base import EventVulnerability
from episimmer.world import World
[docs]class SimpleEventVulnerability(EventVulnerability):
r"""
This class implements the Simple Event Vulnerability module. This class scores the event based on the
severity of outbreak after locking down all other events. Multiple simulations are run by randomly selecting an
event. The event score is calculated by finding the running average of scores for each run.
Inherits :class:`~episimmer.vulnerability_detection.base.EventVulnerability` class.
.. math::
Event\ Score = \frac{Total\ Agents\ Infected}{Total\ Agents}
The Algorithm Parameter File in the vd_config file must contain the parameters 'states', 'num_runs' and
'event_identifier'.
'states' refers to the states in the simulation that are infectious or post-infectious
'num_runs' refers to the total number of simulation runs
'event_identifier' refers to the name of the event identifier to differentiate events
An example of the vd_config.txt file and the algorithm parameter file (parameter.json) is given below
.. code-block:: text
:caption: vd_config.txt
:linenos:
VD Target <Event>
VD Algorithm <SimpleEventVulnerability>
Algorithm Parameter File <parameter.json>
Pre Processing <>
Post Processing <>
Output Mode <Default>
.. code-block:: json
:caption: parameter.json
:linenos:
{
"states":["Infected","Recovered"],
"num_runs":1000,
"event_identifier":"Id"
}
The event identifier will be defined in an individual event file.
.. code-block:: text
:caption: four_event.txt
:linenos:
2
Id:Location Index:Agents
4:1:12,17,9,7,19,11,4,1,13,3,16,2,8,10
5:0:14,2,9,10,4,19,3,18,11,13,15,16,7,5,6,17,0
Args:
world_obj: World object of simulation
parameter_dict: Dictionary of parameters relevant to the algorithm
"""
def __init__(self, world_obj: World,
parameter_dict: Dict[str, Union[List[str], int, str]]):
super().__init__(world_obj)
self.states: List[str] = parameter_dict['states']
self.num_runs: int = parameter_dict['num_runs']
self.event_identifier: str = parameter_dict['event_identifier']
self.selected_event: Union[str, None] = None
self.event_scores: Dict[str, float] = {}
self.event_counts: Dict[str, int] = {}
self.events: List[str] = self.get_events()
self.init_scores()
[docs] def init_scores(self) -> None:
"""
Initialises scores and counts for all event with value 0.0 and 0 respectively.
"""
for event in self.events:
self.event_scores[event] = 0.0
self.event_counts[event] = 0
[docs] def get_events(self) -> List[str]:
"""
Returns list of events in the simulation as event identifiers.
Returns:
List of event identifiers
"""
agents_obj = ReadAgents(self.world_obj.agents_filename,
self.world_obj.config_obj)
locations_obj = ReadLocations(self.world_obj.locations_filename,
self.world_obj.config_obj)
one_time_event_obj = ReadOneTimeEvents(
self.world_obj.one_time_event_file)
for event_files_list in self.world_obj.event_files_list:
if event_files_list:
for events_filename in event_files_list:
ReadEvents(events_filename, self.world_obj.config_obj,
locations_obj, agents_obj)
for time_step in range(self.world_obj.config_obj.time_steps):
one_time_event_obj.populate_one_time_events(
self.world_obj.config_obj, locations_obj, agents_obj,
time_step)
events_id_list = [
event[self.event_identifier]
for location in locations_obj.locations.values()
for event in location.events if self.event_identifier in event
]
events_id_list = list(dict.fromkeys(events_id_list))
return events_id_list
[docs] def select_event(self) -> List[EventLockdown]:
"""
Selects a random event and locks down the rest of the events.
Returns:
List of EventLockdown objects corresponding to all events other than the selected event
"""
self.selected_event = random.choice(self.events)
self.event_counts[self.selected_event] += 1
events_to_remove = [
event for event in self.events if event != self.selected_event
]
return [
EventLockdown(self.event_identifier, events_to_remove,
lambda x: True)
]
[docs] def update_event_scores(self, end_state: Dict[str, List[int]]) -> None:
"""
Function to update the event scores.
Args:
end_state: Dictionary mapping states to time step wise population
"""
infected_agents = sum(end_state[state][-1] for state in self.states)
total_agents = sum(end_state[state][-1] for state in end_state)
event_score = infected_agents / total_agents
if self.selected_event:
self.event_scores[self.selected_event] += 1.0 / self.event_counts[
self.selected_event] * (event_score -
self.event_scores[self.selected_event])
[docs] def one_run(self) -> None:
"""
Executes a single run of the detection module.
"""
Time.new_world()
time_steps = self.world_obj.config_obj.time_steps
agents_obj = ReadAgents(self.world_obj.agents_filename,
self.world_obj.config_obj)
locations_obj = ReadLocations(self.world_obj.locations_filename,
self.world_obj.config_obj)
one_time_event_obj = ReadOneTimeEvents(
self.world_obj.one_time_event_file)
policy_list = self.select_event() if self.events else []
sim_obj = Simulate(self.world_obj.config_obj, self.world_obj.model,
policy_list, agents_obj, locations_obj)
sim_obj.on_start_simulation()
for current_time_step in range(time_steps):
sim_obj.on_start_time_step(
self.world_obj.interaction_files_list,
self.world_obj.event_files_list,
self.world_obj.probabilistic_interaction_files_list,
one_time_event_obj)
sim_obj.handle_time_step_for_all_agents()
sim_obj.end_time_step()
Time.increment_current_time_step()
end_state = sim_obj.end_simulation()
self.update_event_scores(end_state)
[docs] def run_detection(self) -> None:
"""
Function to run the complete detection module.
"""
for i in range(self.num_runs):
if (i + 1) % (self.num_runs / 10) == 0:
print('Iteration running: ', i + 1)
self.one_run()