-
Notifications
You must be signed in to change notification settings - Fork 2
/
SetupFunctions.py
153 lines (135 loc) · 7.65 KB
/
SetupFunctions.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
import math
import random
from scipy.special import erfinv
def sample_point_fn(cb, param_names, param_values):
tags ={}
required_inputs = ['META_Vaccination_Threshold', 'META_Fraction_Meeting', 'META_campaign_coverage', 'META_Dropout']
if not set(required_inputs).issubset(set(param_names)):
raise ValueError('All of ' + ', '.join(required_inputs) + ' must be inputs to sample_point_fn')
# Setup some baseline parameters, but allow them to be overwritten afterwards by inputs to this function
params_dict = Setup_Base_Parameters()
for name, value in zip(param_names, param_values):
params_dict[name] = value
RI_Vacc_Setup(cb, params_dict['META_Vaccination_Threshold'], params_dict['META_Fraction_Meeting'],
params_dict['META_Dropout'], tags)
SIA_Coverage_setup(cb, params_dict['META_campaign_coverage'])
#Now I am through the required parameters
for param, value in params_dict.items():
if param.startswith('META'):
tags = MetaParameterHandler(cb, param, value, tags)
else:
cb.set_param(param, value)
tags[param] = value
return tags
def Setup_Base_Parameters():
# Set some defaults here, but allow them to be overwritten by inputs to the function
params_dict = dict()
params_dict['Base_Population_Scale_Factor'] = 0.075
params_dict['x_Birth'] = 0.98
params_dict['x_Local_Migration'] = 1.0
params_dict['x_Air_Migration'] = 1.0
params_dict['Base_Infectivity'] = 0.6 # R0 of 12 in rural settings
params_dict['Population_Density_Infectivity_Correction'] = 'URBAN_RURAL_INFECTIVITY'
params_dict['Rural_Infectivity_Multiplier'] = 1.0
params_dict['Urban_Infectivity_Multiplier'] = 2.0 # R0 of 24 in urban settings
params_dict['Relative_Sample_Rate_Immune'] = 0.02
params_dict['Infectivity_Scale_Type'] = 'ANNUAL_BOXCAR_FUNCTION'
params_dict['Infectivity_Boxcar_Forcing_Amplitude'] = 0.2
params_dict['Infectivity_Boxcar_Forcing_Start_Time'] = 30
params_dict['Infectivity_Boxcar_Forcing_End_Time'] = 120
params_dict['logLevel_Node'] = "ERROR"
params_dict['logLevel_default'] = "ERROR"
params_dict['logLevel_JsonConfigurable'] = "ERROR"
params_dict['logLevel_StandardEventCoordinator'] = "ERROR"
params_dict['logLevel_Memory'] = "ERROR"
params_dict['Enable_Abort_Zero_Infectivity'] = 1
params_dict['Simulation_Timestep'] = 1.0
params_dict['Spatial_Output_Days_To_Accumulate'] = 30
# mAB_prfiles = [(150, 50), (90, 30), (120, 51)]
params_dict['Maternal_Sigmoid_HalfMaxAge'] = 150
params_dict['Maternal_Sigmoid_SteepFac'] = 50
params_dict['Maternal_Sigmoid_SusInit'] = 0.00
params_dict['Simulation_Duration'] = 9125
return params_dict
def SIA_Coverage_setup(cb, campaign_coverage):
demog = cb.demog_overlays['demographics.json']
#Logic - if campaign coverage is <0, there are no SIAs
#if campaign coverage>=0, there are campaigns that reach 100% of RI children,
#plus "campaign_coverage" fraction of the missed kids.
demographicCov = 0.0 if campaign_coverage < 0 else 1.0
if campaign_coverage <= 0:
campaign_coverages = [0.0]*len(demog['Nodes'])
else:
LN_sig = 0.1 + 2.4 * (1 - random.random() ** (2.0 / 3.0))
LN_mu = math.log(campaign_coverage / (1 - campaign_coverage)) - math.sqrt(2) * LN_sig * erfinv(1 - 2 * 0.5)
tmp = [math.exp(LN_mu + LN_sig*random.gauss(0, 1)) for i in range(len(demog['Nodes']))]
campaign_coverages = [t/(1+t) for t in tmp]
for event in cb.campaign.Events:
if event.Event_Name.startswith('SIA'):
event.Event_Coordinator_Config.Demographic_Coverage = demographicCov
if event.Event_Name == 'SIAs - SIAOnly Group':
event.Event_Coordinator_Config.Coverage_By_Node = []
for ii in range(len(demog['Nodes'])):
event.Event_Coordinator_Config.Coverage_By_Node.append(
[demog['Nodes'][ii]['NodeID'], campaign_coverages[ii]])
def RI_Vacc_Setup(cb, threshold, fraction_meeting, Dropout, tags):
# RI Vaccination is all set up in the demographics file using individual properties
# Draw a random variance and construct the coverages from a logitnormal distribution
#LN_sig = 0.1 + 1.4 * (1 - random.random() ** (2.0 / 3.0)) # slightly bias variance down relative to uniform
LN_sig = 4*random.random()
LN_mu = math.log(threshold / (1 - threshold)) - math.sqrt(2) * LN_sig * erfinv(1 - 2 * fraction_meeting)
# I need to make sure that this is writing correctly - assigning dictionaries in this way in a loop may be tricky.
demog = cb.demog_overlays['demographics.json']
district_names = [node['dot_name'].split(':')[2] for node in demog['Nodes']]
unique_district_names = set(district_names)
district_coverages = {}
for name in unique_district_names:
tmp = math.exp(LN_mu + LN_sig * random.gauss(0, 1))
district_coverages[name] = tmp / (1 + tmp)
tags['LN_sig'] = LN_sig
tags['LN_mu'] = LN_mu
for node in demog['Nodes']:
distcov = district_coverages[node['dot_name'].split(':')[2]]
# Add a little ward-level noise around the district mean
tmp = math.log(distcov / (1 - distcov))
tmp2 = math.exp(tmp + 0.2 * random.gauss(0, 1))
wardcov = tmp2 / (1 + tmp2)
node['IndividualProperties'] = []
node['IndividualProperties'].append(demog['Defaults']['IndividualProperties'][0].copy())
node['IndividualProperties'][0]['Initial_Distribution'] = [(1-Dropout)*wardcov, (Dropout)*wardcov, 1-wardcov]
cb.demog_overlays['demographics.json'] = demog
def MetaParameterHandler(cb, param, value, tags):
#A place to handle all of the various metaparameters that may arise
if param == 'META_Migration':
cb.set_param('x_Local_Migration', value)
cb.set_param('x_Air_Migration', value)
tags['x_Local_Migration'] = value
tags['x_Air_Migration'] = value
if param == 'META_MCV1Days':
for event in cb.campaign.Events:
if event.Event_Name == 'MCV1':
event.Event_Coordinator_Config.Intervention_Config.Actual_IndividualIntervention_Config.Delay_Period_Mean = value
event.Event_Coordinator_Config.Intervention_Config.Actual_IndividualIntervention_Config.Delay_Period_Std_Dev = value/6.0
tags['MCV1_Dose_Days'] = value
if param == 'META_MCV2Days':
for event in cb.campaign.Events:
if event.Event_Name == 'MCV2':
event.Event_Coordinator_Config.Intervention_Config.Actual_IndividualIntervention_Config.Delay_Period_Mean = value
event.Event_Coordinator_Config.Intervention_Config.Actual_IndividualIntervention_Config.Delay_Period_Std_Dev = value/6.0
tags['MCV2_Dose_Days'] = value
if param == 'META_MaB_Profile':
mAb_profiles = {'Long': [150, 50], 'Short': [90, 30], 'Mix': [120, 51]}
cb.set_param('Maternal_Sigmoid_HalfMaxAge', mAb_profiles[value][0])
cb.set_param('Maternal_Sigmoid_SteepFac', mAb_profiles[value][1])
tags['Maternal_Antibody_Profile'] = value
tags['Maternal_Sigmoid_HalfMaxAge'] = mAb_profiles[value][0]
tags['Maternal_Sigmoid_SteepFac'] = mAb_profiles[value][1]
if param == 'META_Timesteps':
cb.set_param('Simulation_Timestep', int(value))
cb.set_param('Spatial_Output_Days_To_Accumulate', int(30/value))
for event in cb.campaign.Events:
event.Event_Coordinator_Config.Timesteps_Between_Repetitions = max(int(1), int(
event.Event_Coordinator_Config.Timesteps_Between_Repetitions/value))
tags['Simulation_Timestep'] = value
tags['Spatial_Output_Days_To_Accumulate'] = 30/value
return tags