episimmer.policy
- class Policy(policy_type)[source]
Class for implementing an Agent policy.
- Parameters
policy_type (str) – Type of Agent Policy
- reset(agents, locations, model, policy_index)[source]
Resets the policy for a new world.
- Parameters
agents (ValuesView[episimmer.agent.Agent]) – Collection of
Agentobjectslocations (ValuesView[episimmer.location.Location]) – Collection of
Locationobjectsmodel (episimmer.model.BaseModel) – Disease model specified by the user
policy_index (int) – Policy index passed to differentiate policies
- Return type
None
- enact_policy(time_step, agents, locations, model, policy_index)[source]
Executes a policy for the given time step.
- Parameters
time_step (int) – Current time step
agents (Dict[str, episimmer.agent.Agent]) – Dictionary mapping from agent indices to
Agentobjectslocations (ValuesView[episimmer.location.Location]) – Collection of
Locationobjectsmodel (episimmer.model.BaseModel) – Disease model specified by the user
policy_index (int) – Policy index passed to differentiate policies
- Return type
None
- post_policy(time_step, agents, locations, model, policy_index)[source]
Post policy procedure used for policies that require the interactions and events that are going to take place in the current time step.
- Parameters
time_step (int) – Current time step
agents (Dict[str, episimmer.agent.Agent]) – Dictionary mapping from agent indices to
Agentobjectslocations (ValuesView[episimmer.location.Location]) – Collection of
Locationobjectsmodel (episimmer.model.BaseModel) – Disease model specified by the user
policy_index (int) – Policy index passed to differentiate policies.
- Return type
None
- class AgentPolicy(policy_type)[source]
Class for implementing an Agent policy.
- Parameters
policy_type (str) – Type of Agent Policy
- update_agent_policy_history(agent, history_value)[source]
Updates the agent policy history list for the current policy type.
- Parameters
agent (episimmer.agent.Agent) – Instance of
Agenthistory_value (object) – Value to be appended to the history list
- Return type
None
- get_agent_policy_history(agent)[source]
Returns the agent policy history list for the current policy type.
- Parameters
agent (episimmer.agent.Agent) – Instance of
Agent- Returns
The agent policy history list for the current policy type
- Return type
object
- update_agent_policy_state(agent, new_state_value)[source]
Updates the agent policy state for the current policy type.
- Parameters
agent (episimmer.agent.Agent) – Instance of
Agentnew_state_value (object) – Value to be set as the policy state
- Return type
None
- get_agent_policy_state(agent)[source]
Returns the agent policy state for the current policy type.
- Parameters
agent (episimmer.agent.Agent) – Instance of
Agent- Returns
The agent policy state for the current policy type
- Return type
object
- class EventPolicy(policy_type)[source]
Class for implementing an Event policy.
- Parameters
policy_type (str) – Type of Event Policy
Lockdown Policy API
Base class for implementing the lockdown policy for agents. |
|
Class for implementing a lockdown policy common for all agents. |
|
Class for implementing the lockdown policy for agents based on a fixed attribute of the agent. |
|
Class for implementing the lockdown policy for agents taking into account their test results. |
|
Base class for implementing the lockdown policy for events. |
|
Class for implementing the lockdown policy for events based on a fixed attribute of the event. |
- class AgentLockdownPolicy(do_lockdown_fn, p)[source]
Base class for implementing the lockdown policy for agents. Inherits
AgentPolicyclass.- Parameters
do_lockdown_fn (Callable) – User-defined function to specify which time step(s) to enforce lockdown in
p (float) – Probability of agent to contribute and receive infection from any source of infection under lockdown
- lockdown_agent(agent)[source]
Updates the agent’s probability to contribute and receive infection.
- Parameters
agent (episimmer.agent.Agent) – Current agent
- Return type
None
- class FullLockdown(do_lockdown_fn, p=0.0)[source]
Class for implementing a lockdown policy common for all agents. Inherits
AgentLockdownPolicyclass.An example of a GeneratePolicy.py file illustrating full lockdown policy where all agents are lockdown on alternate days is given below
1from episimmer.policy import lockdown_policy 2 3def generate_policy(): 4 policy_list=[] 5 6 def lockdown_fn(time_step): 7 if time_step % 2 == 0: 8 return True 9 10 return False 11 12 policy_list.append(lockdown_policy.FullLockdown(lockdown_fn)) 13 14 return policy_list
- Parameters
do_lockdown_fn (Callable) – User-defined function to specify which time step(s) to enforce lockdown in
p (float) – Probability of agent to contribute and receive infection from any source of infection under lockdown
- enact_policy(time_step, agents, locations, model=None, policy_index=None)[source]
If lockdown policy is enforced in the current time step, it restricts all agents from receiving and contributing disease.
- Parameters
time_step (int) – Current time step
agents (Dict[str, episimmer.agent.Agent]) – Dictionary mapping from agent indices to
Agentobjectslocations (ValuesView[episimmer.location.Location]) – Collection of
Locationobjectsmodel (Optional[episimmer.model.BaseModel]) – Disease model specified by the user
policy_index (Optional[int]) – Policy index passed to differentiate policies
- Return type
None
- class AgentLockdown(attribute, value_list, do_lockdown_fn, p=0.0)[source]
Class for implementing the lockdown policy for agents based on a fixed attribute of the agent. Inherits
AgentLockdownPolicyclass.An example of a GeneratePolicy.py file illustrating an agent lockdown policy where agents are lockdown based on their Grade attribute
1from episimmer.policy import lockdown_policy 2 3def generate_policy(): 4 policy_list=[] 5 6 def lockdown_fn(time_step): 7 return True 8 9 policy_list.append(lockdown_policy.AgentLockdown('Grade',['Grade 1'],lockdown_fn)) 10 11 return policy_list
- Parameters
attribute (str) – Parameter (attribute) type of agents
value_list (List[str]) – List of attribute values of agents
do_lockdown_fn (Callable) – User-defined function to specify which time step(s) to enforce lockdown in
p (float) – Probability of agent to contribute and receive infection from any source of infection under lockdown
- enact_policy(time_step, agents, locations, model=None, policy_index=None)[source]
If lockdown policy is enforced in the current time step, it restricts a subset of agents from receiving and contributing disease. The agents selected are based on an agent attribute and the value list for that attribute.
- Parameters
time_step (int) – Current time step
agents (Dict[str, episimmer.agent.Agent]) – Dictionary mapping from agent indices to
Agentobjectslocations (ValuesView[episimmer.location.Location]) – Collection of
Locationobjectsmodel (Optional[episimmer.model.BaseModel]) – Disease model specified by the user
policy_index (Optional[int]) – Policy index passed to differentiate policies
- Return type
None
- class TestingBasedLockdown(do_lockdown_fn, lockdown_period, contact_tracing=False, p=0.0)[source]
Class for implementing the lockdown policy for agents taking into account their test results. This policy also handles locking down contacts of positively tested agents. Inherits
AgentLockdownPolicyclass.An example of a GeneratePolicy.py file illustrating locking down positively tested agents for a period of 10 days is given below
1from episimmer.policy import lockdown_policy, testing_policy 2 3def generate_policy(): 4 policy_list=[] 5 6 Normal_Test = testing_policy.TestPolicy(lambda x:60) 7 Normal_Test.add_machine('Simple_Machine', 200, 0.0, 0.0, 0, 50, 3, 2) 8 Normal_Test.set_register_agent_testtube_func(Normal_Test.random_testing()) 9 policy_list.append(Normal_Test) 10 11 ATP = lockdown_policy.TestingBasedLockdown(lambda x:True,10) 12 policy_list.append(ATP) 13 14 return policy_list
An example of a GeneratePolicy.py file illustrating locking down positively tested agents along with their contacts for a period of 2 days is given below
1from episimmer.policy import (contact_tracing_policy, lockdown_policy, 2 testing_policy) 3 4def generate_policy(): 5 policy_list=[] 6 Normal_Test = testing_policy.TestPolicy(lambda x:7) 7 Normal_Test.add_machine('Simple_Machine', 200, 0.0, 0.0, 0, 50, 3, 2) 8 Normal_Test.set_register_agent_testtube_func(Normal_Test.random_testing()) 9 policy_list.append(Normal_Test) 10 11 CT_object = contact_tracing_policy.CTPolicy(7) 12 policy_list.append(CT_object) 13 14 Lockdown_object = lockdown_policy.TestingBasedLockdown(lambda x:True, 2, True) 15 policy_list.append(Lockdown_object) 16 17 return policy_list
- Parameters
do_lockdown_fn (Callable) – User-defined function to specify which time steps to enforce lockdown in
lockdown_period (int) – Number of time steps for which an agent has to lock down
contact_tracing (bool) – Boolean specifying whether lockdown for contacts of positively tested agents is enabled or not
p (float) – Probability of agent to contribute and receive infection from any source of infection under lockdown
- enact_policy(time_step, agents, locations, model=None, policy_index=None)[source]
If lockdown policy is enforced in the current time step, it restricts positively tested agents from receiving and contributing disease. If contact tracing is enabled, contacts of positively tested agents are also lockdown.
- Parameters
time_step (int) – Current time step
agents (Dict[str, episimmer.agent.Agent]) – Dictionary mapping from agent indices to
Agentobjectslocations (ValuesView[episimmer.location.Location]) – Collection of
Locationobjectsmodel (Optional[episimmer.model.BaseModel]) – Disease model specified by the user
policy_index (Optional[int]) – Policy index passed to differentiate policies
- Return type
None
- lockdown_positive_agents(agents, time_step)[source]
Locks down positively tested agents. If contact tracing is enabled, contacts of positively tested agents are lockdown for a fixed time period. Each contact will save this period as a scheduled time left for lockdown and this value is reduced each time step.
- Parameters
agents (Dict[str, episimmer.agent.Agent]) – Dictionary mapping from agent indices to
Agentobjectstime_step (int) – Current time step
- Return type
None
- lockdown_contacts(agents)[source]
Locks down a contact if its scheduled time left for lockdown is greater than 0.
- Parameters
agents (Dict[str, episimmer.agent.Agent]) – Collection of
Agentobjects.- Return type
None
- class EventLockdownPolicy(do_lockdown_fn, p)[source]
Base class for implementing the lockdown policy for events. Inherits
EventPolicyclass.- Parameters
do_lockdown_fn (Callable) – User-defined function to specify which time step(s) to enforce lockdown in
p (float) – Probability of an event occurring during lockdown
- class EventLockdown(attribute, value_list, do_lockdown_fn, p=0.0)[source]
Class for implementing the lockdown policy for events based on a fixed attribute of the event. Inherits
EventLockdownPolicyclass.An example of a GeneratePolicy.py file illustrating Event lockdown policy where events are lockdown based on an Event attribute. Here, Events of Type - Low Priority are lockdown.
1from episimmer.policy import lockdown_policy, testing_policy 2 3def generate_policy(): 4 policy_list=[] 5 6 event_lockdown = lockdown_policy.EventLockdown('Type', ['Low Priority'], lambda x: True) 7 policy_list.append(event_lockdown) 8 9 return policy_list
- Parameters
attribute (str) – Parameter (attribute) type of events
value_list (List[str]) – List of attribute values of events
do_lockdown_fn (Callable) – User-defined function to specify which time step(s) to enforce lockdown in
p (float) – Probability of an event occurring during lockdown
- enact_policy(time_step, agents, locations, model=None, policy_index=None)[source]
If lockdown policy is enforced in the current time step, it restricts a subset of events from occurring. The events selected are based on an event attribute and the value list for that attribute.
- Parameters
time_step (int) – Current time step
agents (Dict[str, episimmer.agent.Agent]) – Dictionary mapping from agent indices to
Agentobjectslocations (ValuesView[episimmer.location.Location]) – Collection of
Locationobjectsmodel (Optional[episimmer.model.BaseModel]) – Disease model specified by the user
policy_index (Optional[int]) – Policy index passed to differentiate policies
- Return type
None
Testing Policy API
Class for implementing the testing policy. |
|
Class for a Testtube. |
|
Class for a Testing Machine. |
|
Class for a Test Result. |
- class TestPolicy(agents_per_step_fn)[source]
Class for implementing the testing policy. Inherits
AgentPolicyclass.Note that for disease control, we require locking down agents that are positive, thus we have included a lockdown policy in the examples below.
An example of a GeneratePolicy.py file illustrating normally testing random agents (and locking down positively tested agents) is given below.
1from episimmer.policy import lockdown_policy, testing_policy 2 3def generate_policy(): 4 policy_list=[] 5 6 Normal_Test = testing_policy.TestPolicy(lambda x:60) 7 Normal_Test.add_machine('Simple_Machine', 200, 0.0, 0.0, 0, 50, 3, 2) 8 Normal_Test.set_register_agent_testtube_func(Normal_Test.random_testing(1, 1)) 9 policy_list.append(Normal_Test) 10 11 ATP = lockdown_policy.TestingBasedLockdown(lambda x:True,10) 12 policy_list.append(ATP) 13 14 return policy_list
An example of a GeneratePolicy.py file illustrating pool testing random agents with (NAPT, NTPA) = (3,2) (and locking down positively tested agents) is given below
1from episimmer.policy import lockdown_policy, testing_policy 2 3def generate_policy(): 4 policy_list=[] 5 6 Normal_Test = testing_policy.TestPolicy(lambda x:60) 7 Normal_Test.add_machine('Simple_Machine', 200, 0.0, 0.0, 0, 50, 3, 2) 8 Normal_Test.set_register_agent_testtube_func(Normal_Test.random_testing(3, 2)) 9 policy_list.append(Normal_Test) 10 11 ATP = lockdown_policy.TestingBasedLockdown(lambda x:True,10) 12 policy_list.append(ATP) 13 14 return policy_list
An example of a GeneratePolicy.py file illustrating normally testing random agents along with testing their contacts in case they are positive (and locking down positively tested agents) is given below. Here, we need to also include a contact tracing policy to save contacts each time step.
1from episimmer.policy import (contact_tracing_policy, lockdown_policy, 2 testing_policy) 3 4 5def generate_policy(): 6 policy_list=[] 7 Normal_Test1 = testing_policy.TestPolicy(lambda x: 2) 8 Normal_Test1.add_machine('Simple_Machine', 200, 0.0, 0.0, 0, 50, 2, 2) 9 Normal_Test1.set_register_agent_testtube_func(Normal_Test1.random_testing(1, 1)) 10 policy_list.append(Normal_Test1) 11 12 Normal_Test2 = testing_policy.TestPolicy(lambda x: 2) 13 Normal_Test2.add_machine('Simple_Machine', 200, 0.0, 0.0, 0, 50, 2, 2) 14 Normal_Test2.set_register_agent_testtube_func(Normal_Test2.contact_testing(1, 1)) 15 policy_list.append(Normal_Test2) 16 17 CT_object = contact_tracing_policy.CTPolicy(7) 18 policy_list.append(CT_object) 19 20 Lockdown_object = lockdown_policy.TestingBasedLockdown(lambda x:1, 2) 21 policy_list.append(Lockdown_object) 22 23 return policy_list
- Parameters
agents_per_step_fn (Callable) – User-defined function to specify the number of agents to test per time step
- reset(agents=None, locations=None, model=None, policy_index=None)[source]
Resets statistics, ready queue and all the machines for a new world.
- Parameters
agents (Optional[ValuesView[episimmer.agent.Agent]]) – Collection of
Agentobjectslocations (Optional[ValuesView[episimmer.location.Location]]) – Collection of
Locationobjectsmodel (Optional[episimmer.model.BaseModel]) – Disease model specified by the user
policy_index (Optional[int]) – Policy index passed to differentiate policies
- Return type
None
- enact_policy(time_step, agents, locations, model, policy_index=None)[source]
Executes testing policy for the given time step.
- Parameters
time_step (int) – Time step in which the policy is enacted
agents (Dict[str, episimmer.agent.Agent]) – Dictionary mapping from agent indices to
Agentobjectslocations (ValuesView[episimmer.location.Location]) – Collection of
Locationobjectsmodel (episimmer.model.BaseModel) – Disease model specified by the user
policy_index (Optional[int]) – Policy index passed to differentiate policies
- Return type
None
- zero_turnaround_time_func(time_step)[source]
When turnaround time = 0, results are populated in the same time step. This function handles that scenario.
- Parameters
time_step (int) – Current time step
- Return type
None
- add_machine(machine_name, cost, false_positive_rate, false_negative_rate, turnaround_time, capacity, valid_period, num=1)[source]
This function enables the user to add a machine in the
Generate_policy.pyfile. A machine must be defined as it performs the testing procedure on testtubes. It defines parameters of the testing policy.1Normal_Test.add_machine('Simple_Machine', 200, 0.0, 0.0, 0, 50, 2, 2)
- Parameters
machine_name (str) – Name of machine
cost (int) – Cost for a single test in the machine
false_positive_rate (float) – False positive rate of the machine
false_negative_rate (float) – False negative rate of the machine
turnaround_time (int) – Time taken for a test result
capacity (int) – Capacity of the machine for tests
valid_period (int) – Number of time steps the test is considered to be valid
num (int) – Number of instances of this machine
- Return type
None
- set_register_agent_testtube_func(fn)[source]
Registers the function that determines how agents are mapped to testtubes. The user must specify one of the following functions in the
Generate_policy.pyfile.The example below illustrates the use of both testing methods.
1Normal_Test1.set_register_agent_testtube_func(Normal_Test1.random_testing(1, 1)) 2Normal_Test2.set_register_agent_testtube_func(Normal_Test2.contact_testing(1, 1))
- Parameters
fn (Callable) – Function that determines the type of testing to be performed
- Return type
None
- initialize_statistics_logs(time_step)[source]
Initializes statistics dictionary. It contains global information of all tests.
- Parameters
time_step (int) – Current time step
- Return type
None
- initialize_process_logs(time_step)[source]
Initializes statistics dictionary for each machine. It contains machine-level information.
- Parameters
time_step (int) – Current time step
- Return type
None
- new_time_step(time_step)[source]
Initializes statistics of the testing policy. Resets the current test tubes and the number of agents to test in the current time step.
- Parameters
time_step (int) – Current time step
- Return type
None
- populate_test_queue(agents_to_test, num_agents_per_testtube, num_testtubes_per_agent, time_step)[source]
Populates the testing ready queue with fully filled testtubes containing agent samples. This method can handle both regular testing and pool testing using the parameters \(NAPT\) (number of agents per testtube) and \(NTPA\) (number of testtubes per agent) passed from the function defining the mapping from agents to testtubes. The number of testtubes \(N_T\) required follow the formula -
\[N_T = \lfloor \frac{NTPA \times N_A + NAPT - 1}{NAPT} \rfloor\]where \(N_A\) denotes the number of agents to test.
- Parameters
agents_to_test (List[episimmer.agent.Agent]) – List of
Agentobjects ready for testingnum_agents_per_testtube (int) – Number of agents per testtube
num_testtubes_per_agent (int) – Number of testtubes per agent
time_step (int) – Current time step
- Return type
None
- full_random_testing(num_agents_per_testtube, num_testtubes_per_agent, only_symptomatic, attribute, value_list, agents, time_step, model)[source]
Agents are first selected for testing and added to a list based on the number of agents to test in the current time step, agent parameters (if given) and symptomatic states (if set to True). Then, the test ready queue is populated.
- Parameters
num_agents_per_testtube (int) – Number of agents per testtube (NAPT)
num_testtubes_per_agent (int) – Number of testtubes per agent (NTPA)
only_symptomatic (bool) – Choose whether to only select symptomatic agents or not (If set to True, you must have symptomatic states set in
UserModel.py)attribute (Optional[str]) – Parameter (attribute) type of agents
value_list (List[str]) – List of attribute values of agents
agents (Dict[str, episimmer.agent.Agent]) – Collection of
Agentobjectstime_step (int) – Current time step
model (episimmer.model.BaseModel) – Disease model specified by the user
- Return type
None
- random_testing(num_agents_per_testtube=1, num_testtubes_per_agent=1, only_symptomatic=False, attribute=None, value_list=[])[source]
This function can be used by the user in
Generate_policy.pyto test random agents. This function can handle normal or pool testing. Normal testing refers to testing a single agent once i.e. A single agent’s sample present in a single testtube. Pool testing refers to having multiple agents in a testtube defined by the num_agents_per_testtube parameter and multiple testtubes for an agent defined by the num_testtubes_per_agent parameter. If symptomatic states are defined in the disease model in theUserModel.pyfile, then you may also only test symptomatic agents. This function returns a partial function offull_random_testing().An example of a GeneratePolicy.py file illustrating normally testing and pool testing random agents (and locking down positively tested agents) is given below.
1from episimmer.policy import lockdown_policy, testing_policy 2 3def generate_policy(): 4 policy_list=[] 5 6 Normal_Test = testing_policy.TestPolicy(lambda x:60) 7 Normal_Test.add_machine('Simple_Machine', 200, 0.0, 0.0, 0, 50, 3, 2) 8 Normal_Test.set_register_agent_testtube_func(Normal_Test.random_testing(1, 1)) 9 policy_list.append(Normal_Test) 10 11 Pool_Testing = testing_policy.TestPolicy(lambda x:150) 12 Pool_Testing.add_machine('Simple_Machine', 200, 0.0, 0.0, 0, 50, 3, 2) 13 Pool_Testing.set_register_agent_testtube_func(Pool_Testing.random_testing(5,2)) 14 policy_list.append(Pool_Testing) 15 16 ATP = lockdown_policy.TestingBasedLockdown(lambda x:True,10) 17 policy_list.append(ATP) 18 19 return policy_list
- Parameters
num_agents_per_testtube (int) – Number of agents per testtube (NAPT)
num_testtubes_per_agent (int) – Number of testtubes per agent (NAPT)
only_symptomatic (bool) – Choose whether to only select symptomatic agents or not (If set to True, you must have symptomatic states set in
UserModel.py)attribute (Optional[str]) – Parameter (attribute) type of agents
value_list (List[str]) – List of attribute values of agents
- Returns
Partial function of
full_random_testing()- Return type
Callable
- full_contact_testing(num_agents_per_testtube, num_testtubes_per_agent, attribute, value_list, agents, time_step, model)[source]
Agents are first checked for positive history of testing and then contacts of the positive agents are selected for testing. They are added to a list based on the number of agents to test in the current time step, agent parameters (if given) and symptomatic states (if set to True). Then, the test ready queue is populated.
- Parameters
num_agents_per_testtube (int) – Number of agents per testtube (NAPT)
num_testtubes_per_agent (int) – Number of testtubes per agent (NTPA)
attribute (Optional[str]) – Parameter (attribute) type of agents
value_list (List[str]) – List of attribute values of agents
agents (Dict[str, episimmer.agent.Agent]) – Collection of
Agentobjectstime_step (int) – Current time step
model (episimmer.model.BaseModel) – Disease model specified by the user
- Return type
None
- contact_testing(num_agents_per_testtube=1, num_testtubes_per_agent=1, attribute=None, value_list=[])[source]
This function can be used by the user in
Generate_policy.pyto test contacts of positive agents. This function can handle normal or pool testing. Normal testing refers to testing a single agent once i.e. A single agent’s sample present in a single testtube. Pool testing refers to having multiple agents in a testtube defined by the num_agents_per_testtube parameter and multiple testtubes for an agent defined by the num_testtubes_per_agent parameter. This function returns a partial function offull_contact_testing().An example of a GeneratePolicy.py file illustrating normally testing random agents along with testing their contacts in case they are positive (and locking down positively tested agents) is given below. Here, we need to also include a contact tracing policy to save contacts each time step.
1from episimmer.policy import (contact_tracing_policy, lockdown_policy, 2 testing_policy) 3 4 5def generate_policy(): 6 policy_list=[] 7 Normal_Test1 = testing_policy.TestPolicy(lambda x: 2) 8 Normal_Test1.add_machine('Simple_Machine', 200, 0.0, 0.0, 0, 50, 2, 2) 9 Normal_Test1.set_register_agent_testtube_func(Normal_Test1.random_testing(1, 1)) 10 policy_list.append(Normal_Test1) 11 12 Normal_Test2 = testing_policy.TestPolicy(lambda x: 2) 13 Normal_Test2.add_machine('Simple_Machine', 200, 0.0, 0.0, 0, 50, 2, 2) 14 Normal_Test2.set_register_agent_testtube_func(Normal_Test2.contact_testing(1, 1)) 15 policy_list.append(Normal_Test2) 16 17 CT_object = contact_tracing_policy.CTPolicy(7) 18 policy_list.append(CT_object) 19 20 Lockdown_object = lockdown_policy.TestingBasedLockdown(lambda x:1, 2) 21 policy_list.append(Lockdown_object) 22 23 return policy_list
- Parameters
num_agents_per_testtube (int) – Number of agents per testtube (NAPT)
num_testtubes_per_agent (int) – Number of testtubes per agent (NAPT)
attribute (Optional[str]) – Parameter (attribute) type of agents
value_list (List[str]) – List of attribute values of agents
- Returns
Partial function of
full_contact_testing()- Return type
Callable
- add_partial_to_ready_queue()[source]
Since the
populate_test_queue()method only populates fully filled testtubes, this function adds partially filled testtubes to the test ready queue.- Return type
None
- register_testtubes_to_machines()[source]
Registers the testtubes to all empty/partially filled non-running machines defined by user.
- Return type
None
- run_tests(model, time_step)[source]
Runs the tests for all non-empty non-running machines.
- Parameters
model (episimmer.model.BaseModel) – Disease model specified by the user
time_step (int) – Current time step
- Return type
None
- populate_results_in_machine(time_step)[source]
Populates the results of each test in the machines (if test is complete).
- Parameters
time_step (int) – Current time step
- Return type
None
- release_results_to_agents(results)[source]
Results are released to the agents and the policy history of the agent is updated.
- Parameters
results (List[episimmer.policy.testing_policy.TestResult]) – Collection of
TestResultobjects- Return type
None
- release_results_to_policy(results, time_step)[source]
Results are released to the policy i.e. stored in the statistics dictionary.
- Parameters
results (List[episimmer.policy.testing_policy.TestResult]) – Collection of
TestResultobjectstime_step (int) – Current time step
- Return type
None
- release_results(time_step)[source]
Results are released to the agents and policy once the machine has been populated with results. The machine is then reset.
- Parameters
time_step (int) – Current time step
- Return type
None
- update_process_logs(time_step)[source]
Machine-level logging done here
- Parameters
time_step (int) – Current time step
- Return type
None
- end_time_step(time_step)[source]
At the end of the time step, process logs are updated and stored as a json.
- Parameters
time_step (int) – Current time step
- Return type
None
- static get_accumulated_test_result(history, last_time_step)[source]
Method to get the most recent test result of an agent. In the case of pool testing, if even one of the pool tests return negative, he is a negatively tested agent, otherwise, if all pool tests return positive, he is a positively tested agent.
- Parameters
history (List[episimmer.policy.testing_policy.TestResult]) – Test history of the agent
last_time_step (int) – Most recent time step in which agent was tested
- Returns
A string either “Positive” or “Negative” representing the most recent test result of the agent
- Return type
str
- static get_agent_test_result(agent, time_step)[source]
Returns the most recent test result of an agent (if it exists).
- Parameters
agent (episimmer.agent.Agent) – Current Agent
time_step (int) – Current time step
- Returns
A string either “Positive” or “Negative” representing the most recent test result of the agent. None if no tests done for the agent.
- Return type
Optional[str]
- static is_agent_test_ready(agent, time_step)[source]
Returns a boolean representing whether an agent can test. If the agent has tested before, he can only test once the validity of the latest test has expired.
- Parameters
agent (episimmer.agent.Agent) – Current Agent
time_step (int) – Current time step
- Returns
A boolean representing agent’s ability to test
- Return type
bool
- class TestTube[source]
Class for a Testtube.
- register_agent(agent, time_step)[source]
Testtube registering an agent.
- Parameters
agent (episimmer.agent.Agent) – Instance of
Agenttime_step (int) – Current time step
- Return type
None
- get_num_agents()[source]
Returns the number of agents in the testtube.
- Returns
Number of agents in testtube
- Return type
int
- set_result(result)[source]
Sets the result of test - Positive or Negative.
- Parameters
result (str) – Result of test
- Return type
None
- set_in_machine(bool_val)[source]
Sets the testtube to be in the machine.
- Parameters
bool_val (bool) – Boolean for in machine
- Return type
None
- class Machine(machine_name, cost, false_positive_rate, false_negative_rate, turnaround_time, capacity, valid_period)[source]
Class for a Testing Machine.
- Parameters
machine_name (str) – Name of machine
cost (int) – Cost for a single test in the machine
false_positive_rate (float) – False positive rate of the machine
false_negative_rate (float) – False negative rate of the machine
turnaround_time (int) – Time taken for a test result
capacity (int) – Capacity of the machine for tests
valid_period (int) – Number of time steps the test is considered to be valid
- is_running()[source]
Returns a boolean indicating whether machine is running.
- Returns
Boolean indicating whether machine is running
- Return type
bool
- is_full()[source]
Returns a boolean indicating whether machine is full.
- Returns
Boolean indicating whether machine is full
- Return type
bool
- is_empty()[source]
Returns a boolean indicating whether machine is completely empty.
- Returns
Boolean indicating whether machine is completely empty
- Return type
bool
- has_empty_results()[source]
Returns a boolean indicating whether machine has no results.
- Returns
Boolean indicating whether machine has no results
- Return type
bool
- reset_machine()[source]
Resets the machine’s list of testtubes, results and sets its availability to True.
- Return type
None
- register_testtube(testtube)[source]
Registers a testtube to the machine. Since each testtube corresponds to a single test, the machine cost accumulates the saved cost value once.
- Parameters
testtube (episimmer.policy.testing_policy.TestTube) – Instance of
TestTube- Return type
None
- run_tests(infected_states, time_step)[source]
Runs the tests for each testtube in the machine.
- Parameters
infected_states (List[str]) – Infected states of the disease model
time_step (int) – Current time step
- Return type
None
- run_single_test(testtube, infected_states)[source]
Runs a single test for a testtube in the machine and saves the result for that testtube.
- Parameters
testtube (episimmer.policy.testing_policy.TestTube) – Instance of
TestTubeinfected_states (List[str]) – Infected states of the disease model
- Return type
None
- populate_machine_results(time_step)[source]
Populates the machine with results for each testtube if the machine has completed running. It also removes the testtube from the machine.
- Parameters
time_step (int) – Current time step
- Return type
None
- run_completed(time_step)[source]
Returns a boolean indicating whether machine has completed running tests.
- Returns
Boolean indicating whether machine has completed running tests
- Parameters
time_step (int) –
- Return type
bool
- save_results(testtube, time_step)[source]
Saves the results for a testtube in the results list.
- Parameters
testtube (episimmer.policy.testing_policy.TestTube) – Instance of
TestTubetime_step (int) – Current time step
- Return type
None
- class TestResult(result, agent, machine_name, time_step, machine_start_step, time_step_done, valid_period)[source]
Class for a Test Result.
- Parameters
result (str) – Result of test for testtube
agent (episimmer.agent.Agent) – Instance of the
Agenttestedmachine_name (str) – Name of machine used for testing
time_step (int) – Time step agent tested
machine_start_step (int) – Time step machine started testing
time_step_done (int) – Time step machine completed test
valid_period (int) – Number of time steps the test is considered to be valid
Vaccination Policy API
Class for implementing the vaccination policy. |
|
Class for Vaccine. |
|
Stores the information regarding the result of a vaccination. |
- class VaccinationPolicy(agents_per_step_fn)[source]
Class for implementing the vaccination policy. Inherits
AgentPolicyclass.An example of a GeneratePolicy.py file illustrating single dose and multi dose vaccination is given below.
1from episimmer.policy import vaccination_policy 2 3def generate_policy(): 4 policy_list=[] 5 6 # Single Dose Vaccination 7 vp1= vaccination_policy.VaccinationPolicy(lambda x: 100) 8 vaccines1 = { 9 'cov_single_dose': {'cost': 40, 'count': 20, 'efficacy': 0.9, 'decay': 40}, 10 'cov_single_dose2': {'cost': 50, 'count': 15, 'efficacy': 0.5, 'decay': 30}, 11 } 12 vp1.add_vaccines(vaccines1, 'Single') 13 vp1.set_register_agent_vaccine_func(vp1.random_vaccination()) 14 policy_list.append(vp1) 15 16 # Multi Dose Vaccination 17 vp2= vaccination_policy.VaccinationPolicy(lambda x: 100) 18 vaccines2 = { 19 'cov_multi_dose': {'cost': 40, 'count': 25, 'efficacy': 0.4, 'decay': [15, 14, 8], 'dose': 3, 'interval': [3, 2]}, 20 'cov_multi_dose2': {'cost': 30, 'count': 40, 'efficacy': 0.7, 'decay': [20, 25, 17, 5], 'dose': 4, 'interval': [12, 26, 14]}, 21 'cov_multi_dose3': {'cost': 30, 'count': 15, 'efficacy': 0.7, 'decay': [8], 'dose': 1, 'interval': []} 22 } 23 vp2.add_vaccines(vaccines2, 'Multi') 24 vp2.set_register_agent_vaccine_func(vp2.multi_dose_vaccination()) 25 policy_list.append(vp2) 26 27 return policy_list
- Parameters
agents_per_step_fn (Callable) – User-defined function to specify the number of agents to vaccinate per time step
- enact_policy(time_step, agents, locations, model=None, policy_index=None)[source]
Executes vaccination policy for the given time step.
- Parameters
time_step (int) – Time step in which the policy is enacted
agents (Dict[str, episimmer.agent.Agent]) – Dictionary mapping from agent indices to
Agentobjectslocations (ValuesView[episimmer.location.Location]) – Collection of
Locationobjectsmodel (Optional[episimmer.model.BaseModel]) – Disease model specified by the user
policy_index (Optional[int]) – Policy index passed to differentiate policies.
- Return type
None
- new_time_step(time_step)[source]
Creates a list in which vaccine objects are added according to the user’s specification. Resets the results of the policy enacted in previous time step and the number of agents to vaccinate in the current time step.
- Parameters
time_step (int) – Current time step
- Return type
None
- add_vaccines(vaccines, dosage='Single')[source]
This function enables the user to add vaccines.
Parameters to be specified for single dose vaccines in the vaccines dict:
cost: Cost of vaccine.
count: Number of vaccine available.
efficacy: Vaccine efficacy.
decay: Number of days of protection offered by the vaccine.
1vp1= vaccination_policy.VaccinationPolicy(lambda x: 100) 2vaccines1 = { 3 'cov_single_dose': {'cost': 40, 'count': 20, 'efficacy': 0.9, 'decay': 40}, 4 'cov_single_dose2': {'cost': 50, 'count': 15, 'efficacy': 0.5, 'decay': 30}, 5} 6vp1.add_vaccines(vaccines1, 'Single')
Parameters to be specified for multi dose vaccines in the vaccines dict:
cost: Cost of vaccine.
count: Number of vaccine available.
efficacy: Vaccine efficacy.
decay: A list of number of days of protection offered by each dose of the vaccine.
dose: Number of doses of the vaccine.
interval: A list specifying minimum days to pass before the administration of the next dose for each dose.
1vp2= vaccination_policy.VaccinationPolicy(lambda x: 100) 2vaccines2 = { 3 'cov_multi_dose': {'cost': 40, 'count': 25, 'efficacy': 0.4, 'decay': [15, 14, 8], 'dose': 3, 'interval': [3, 2]}, 4 'cov_multi_dose2': {'cost': 30, 'count': 40, 'efficacy': 0.7, 'decay': [20, 25, 17, 5], 'dose': 4, 'interval': [12, 26, 14]}, 5 'cov_multi_dose3': {'cost': 30, 'count': 15, 'efficacy': 0.7, 'decay': [8], 'dose': 1, 'interval': []} 6} 7vp2.add_vaccines(vaccines2, 'Multi')
- Parameters
vaccines (Dict[str, Dict[str, Union[int, float, List[int], str]]]) – A dictionary mapping vaccine names to its parameters
dosage (str) – Specifies if the vaccines are either
Singledose orMultidose
- Return type
None
- set_register_agent_vaccine_func(func)[source]
Registers the function that determines the type of vaccination to be performed. The user must specify one of the following functions
1vp1.set_register_agent_vaccine_func(vp1.random_vaccination()) 2vp2.set_register_agent_vaccine_func(vp2.multi_dose_vaccination())
- Parameters
func (Callable) – Function that determines the type of vaccination to be performed
- Return type
None
- full_random_vaccination(attribute, value_list, agents, time_step)[source]
If the number of agents vaccinated is less than the maximum number of agents to vaccinate per time step, for every unvaccinated agent this function randomly chooses a vaccine from the list of vaccines and performs vaccination on the agent. This function is valid only for single dose vaccines.
- Parameters
agents (ValuesView[episimmer.agent.Agent]) – Collection of
Agentobjectstime_step (int) – Current time step
attribute (Optional[str]) – Attribute name of agents
value_list (List[str]) – List of attribute values of agents
- Return type
None
- random_vaccination(attribute=None, value_list=[])[source]
This function can be used by the user in
Generate_policy.pyto specify randomized vaccination to be performed for the agents. This function returns a partial function offull_random_vaccination().An example of a GeneratePolicy.py file illustrating single dose vaccination is given below.
1from episimmer.policy import vaccination_policy 2 3def generate_policy(): 4 policy_list=[] 5 6 vp1= vaccination_policy.VaccinationPolicy(lambda x: 100) 7 vaccines1 = { 8 'cov_single_dose': {'cost': 40, 'count': 20, 'efficacy': 0.9, 'decay': 40}, 9 'cov_single_dose2': {'cost': 50, 'count': 15, 'efficacy': 0.5, 'decay': 30}, 10 } 11 vp1.add_vaccines(vaccines1, 'Single') 12 vp1.set_register_agent_vaccine_func(vp1.random_vaccination()) 13 policy_list.append(vp1) 14 15 return policy_list
- Parameters
attribute (Optional[str]) – Attribute name of agents
value_list (List[str]) – List of attribute values of agents
- Returns
Partial function of
full_random_vaccination()- Return type
Callable
- full_multi_dose_vaccination(attribute, value_list, agents, time_step)[source]
If the number of agents vaccinated is less than the maximum number of agents to vaccinate per time step, for every unvaccinated agent this function randomly chooses a vaccine from the list of vaccines and performs vaccination on the agent, and for every vaccinated agent if it is time for next dose, the next dose of the same vaccine is vaccinated for the agent. This function is valid only for multi dose vaccines.
- Parameters
agents (ValuesView[episimmer.agent.Agent]) – Collection of
Agentobjectstime_step (int) – Current time step
attribute (Optional[str]) – Attribute name of agents
value_list (List[str]) – List of attribute values of agents
- Return type
None
- multi_dose_vaccination(attribute=None, value_list=[])[source]
This function can be used by the user in
Generate_policy.pyto specify multi-dose vaccination to be performed for the agents. This function returns a partial function offull_multi_dose_vaccination().An example of a GeneratePolicy.py file illustrating multi dose vaccination is given below.
1from episimmer.policy import vaccination_policy 2 3def generate_policy(): 4 policy_list=[] 5 6 vp2= vaccination_policy.VaccinationPolicy(lambda x: 100) 7 vaccines2 = { 8 'cov_multi_dose': {'cost': 40, 'count': 25, 'efficacy': 0.4, 'decay': [15, 14, 8], 'dose': 3, 'interval': [3, 2]}, 9 'cov_multi_dose2': {'cost': 30, 'count': 40, 'efficacy': 0.7, 'decay': [20, 25, 17, 5], 'dose': 4, 'interval': [12, 26, 14]}, 10 'cov_multi_dose3': {'cost': 30, 'count': 15, 'efficacy': 0.7, 'decay': [8], 'dose': 1, 'interval': []} 11 } 12 vp2.add_vaccines(vaccines2, 'Multi') 13 vp2.set_register_agent_vaccine_func(vp2.multi_dose_vaccination()) 14 policy_list.append(vp2) 15 16 return policy_list
- Parameters
attribute (Optional[str]) – Attribute name of agents
value_list (List[str]) – List of attribute values of agents
- Returns
Partial function of
full_multi_dose_vaccination()- Return type
Callable
- set_protection(agents)[source]
For every vaccinated agent the protection days offered by the vaccine in agent history is decremented by 1.
- Parameters
agents (ValuesView[episimmer.agent.Agent]) – Collection of
Agentobjects- Return type
None
- populate_results()[source]
Updates agent policy history and state from the list of results.
- Return type
None
- restrict_agents(agents)[source]
Restricts the ability of a vaccinated agent to receive an infection.
- Parameters
agents (ValuesView[episimmer.agent.Agent]) – Collection of
Agentobjects- Return type
None
- class VaccineType(name, cost, decay, efficacy, dosage=None, interval=None)[source]
Class for Vaccine.
- Parameters
name (str) – Vaccine name
cost (int) – Cost of the vaccine
decay (Union[List[int], int]) – Number of days of protection offered by the vaccine, a list of each dose in case of multi-dose vaccine
efficacy (float) – Efficacy of the vaccine
dosage (Optional[int]) – Number of doses of the vaccine, applies only for multi-dose vaccine
interval (Optional[List[int]]) – List specifying minimum days to pass before the administration of the next dose, for each dose of a multi-dose vaccine
- vaccinate(agent, time_step, dose=1)[source]
Administers the specified dose of the current vaccine to the agent. Updates the protection days according to the dose administered.
- Parameters
agent (episimmer.agent.Agent) – Agent to vaccinate
time_step (int) – Time step when the vaccination is performed
dose (int) – The dose of the vaccine to be administered to the agent
- Returns
Result object of vaccination
- Return type
- class VaccineResult(vaccine_name, agent, result, time_step, efficacy, decay_days, current_dose)[source]
Stores the information regarding the result of a vaccination.
- Parameters
vaccine_name (str) – Name of the vaccine used
agent (episimmer.agent.Agent) – Agent that was vaccinated
result (str) – Vaccination result (Successful or Unsuccessful)
time_step (int) – Time step when the vaccination was done
efficacy (float) – Efficacy of the vaccine administered to the agent
decay_days (int) – Number of days of protection offered by the vaccine
current_dose (int) – The dose of the vaccine administered to the agent
Contact Tracing Policy API
- class CTPolicy(num_of_days, attribute=None, value_list=[])[source]
Class built to implement the contact tracing policy. Agent contacts are saved based on all types of interactions in Episimmer. You can optionally choose the agents that save their contacts by passing an agent attribute and a value list. Inherits
AgentPolicyclass.An example of a GeneratePolicy.py file illustrating contact tracing policy is given below. It saves contacts for agents of Type Teacher and Student for a period of 7 and 3 time steps respectively.
1from episimmer.policy import contact_tracing_policy 2 3def generate_policy(): 4 policy_list=[] 5 6 CT_object = contact_tracing_policy.CTPolicy(7, 'Type', ['Teacher']) 7 CT_object2 = contact_tracing_policy.CTPolicy(3, 'Type', ['Student']) 8 9 policy_list.append(CT_object) 10 policy_list.append(CT_object2) 11 12 return policy_list
- Parameters
num_of_days (int) – Number of days to store the agent’s contacts
attribute (Optional[str]) – Parameter (attribute) type of agents
value_list (List[str]) – List of attribute values of agents
- reset(agents, locations, model, policy_index)[source]
Resets all the agents contact tracing policy state for a new world.
- Parameters
agents (ValuesView[episimmer.agent.Agent]) – Collection of
Agentobjectslocations (ValuesView[episimmer.location.Location]) – Collection of
Locationobjectsmodel (episimmer.model.BaseModel) – Disease model specified by the user
policy_index (int) – Policy index passed to differentiate policies.
- Return type
None
- post_policy(time_step, agents, locations, model, policy_index)[source]
Runs the post policy method to save contacts.
- Parameters
time_step (int) – Current time step
agents (Dict[str, episimmer.agent.Agent]) – Dictionary mapping from agent indices to
Agentobjectslocations (ValuesView[episimmer.location.Location]) – Collection of
Locationobjectsmodel (episimmer.model.BaseModel) – Disease model specified by the user
policy_index (int) – Policy index passed to differentiate policies
- Return type
None
- new_time_step(agents, policy_index)[source]
Initialises the contact list for agents for the new time step
- Parameters
agents (Dict[str, episimmer.agent.Agent]) – Dictionary mapping from agent indices to
Agentobjectspolicy_index (int) – Policy index passed to differentiate policies
- Return type
None
- save_interactions(agents, policy_index)[source]
Saving the contacts of the agent accounting individual and probabilistic interactions.
- Parameters
agents (Dict[str, episimmer.agent.Agent]) – Dictionary mapping from agent indices to
Agentobjectspolicy_index (int) – Policy index passed to differentiate policies
- Return type
None
- save_events(agents, locations, policy_index)[source]
Saving the contacts of the agent accounting regular and one time events.
- Parameters
agents (Dict[str, episimmer.agent.Agent]) – Dictionary mapping from agent indices to
Agentobjectslocations (ValuesView[episimmer.location.Location]) – Collection of
Locationobjectspolicy_index (int) – Policy index passed to differentiate policies
- Return type
None
- static reduce_agent_schedule_time(agent_ct_state, policy_index)[source]
If the agent’s scheduled time is positive, this function decrements its scheduled time to lockdown by 1.
- Parameters
agent_ct_state (Dict[int, Dict[str, Union[int, Deque[episimmer.policy.contact_tracing_policy.ContactList]]]]) – State of the contact tracing policy of the agent
policy_index (int) – Policy index passed to differentiate policies
- Return type
None
- static reduce_agents_schedule_time(agents)[source]
Reduces the scheduled time for contacts by 1 across all contact tracing policies. This method must be run only once during a time step, even if multiple policies call this method.
- Parameters
agents (Dict[str, episimmer.agent.Agent]) – Collection of
Agentobjects- Return type
None
- static get_policy_index_list(agent)[source]
Gets the list of policy indices corresponding to contact tracing policies
- Parameters
agent (episimmer.agent.Agent) – Current agent
- Returns
List of policy indices (if they exist)
- Return type
Optional[List[int]]
- static get_contact_list(agent, policy_index)[source]
Gets the contacts saved in an agent’s contact tracing policy state for a particular policy index.
- Parameters
agent (episimmer.agent.Agent) – Agent whose contacts are to be returned
policy_index (int) – Policy index of contact tracing policy
- Returns
All the contacts of the agent saved across all contact tracing policies
- Return type
List[str]
- static set_contacts_schedule_time(agents, contacts, policy_index, time_period)[source]
Sets the time needed to lockdown for contacts of a positive agent to a fixed period. If the scheduled time of the contact is not 0 (contact currently in lockdown), then the time needed to lockdown is not reset.
- Parameters
agents (Dict[str, episimmer.agent.Agent]) – Collection of
Agentobjectscontacts (List[str]) – List of contacts of a positively tested agent
policy_index (int) – Policy index of contact tracing policy
time_period (int) – Time period of lockdown
- Return type
None
- static get_max_schedule_time(agent)[source]
Gets the maximum scheduled time left across contact tracing policies.
- Parameters
agent (episimmer.agent.Agent) – Current agent
- Returns
Maximum schedule time left for lockdown for an agent
- Return type
int