import emod_api.interventions.utils as utils
from emod_api import schema_to_class as s2c
import copy
from typing import List
schema_path=None
old_adhoc_trigger_style=True
cached_sec = None
cached_ce = None
_MAX_AGE=365*125
###
### Generic
###
cached_be = None
[docs]def BroadcastEvent(
camp,
Event_Trigger: str = 'Births'
):
"""
Wrapper function to create and return a BroadcastEvent intervention.
Args:
camp: emod_api.campaign object with schema_path set.
Event_Trigger: A valid trigger/event/signal.
Returns:
ReadOnlyDict: Schema-based smart dictionary representing a new
BroadastEvent intervention ready to be added to a campaign.
"""
if Event_Trigger is None or Event_Trigger == "":
raise ValueError( "BroadcastEvent called with an empty Event_Trigger. Please specify a string." )
global cached_be
if cached_be is None:
global schema_path
schema_path = ( camp.schema_path if camp is not None else schema_path )
cached_be = s2c.get_class_with_defaults( "BroadcastEvent", schema_path )
intervention = copy.deepcopy(cached_be)
intervention.Broadcast_Event = camp.get_send_trigger( Event_Trigger, old=old_adhoc_trigger_style )
return intervention
[docs]def BroadcastEventToOtherNodes(
camp,
Event_Trigger,
Node_Selection_Type="DISTANCE_ONLY",
Max_Distance_To_Other_Nodes_Km=-1,
Include_My_Node=1
):
"""
Wrapper function to create and return a BroadcastEventToOtherNodes intervention.
Args:
camp: emod_api.campaign object with schema_path set.
Event_Trigger: A valid trigger/event/signal.
Node_Selection_Type: TBD.
Max_Distance_To_Other_Nodes_Km: TBD.
Include_My_Node: TBD.
Returns:
ReadOnlyDict: Schema-based smart dictionary representing a new
BroadastEvent intervention ready to be added to a campaign.
"""
if Event_Trigger is None or Event_Trigger == "":
raise ValueError( "BroadcastEventToOtherNodes called with an empty Event_Trigger. Please specify a string." )
global schema_path
schema_path = ( camp.schema_path if camp is not None else schema_path )
intervention = s2c.get_class_with_defaults( "BroadcastEventToOtherNodes", schema_path )
intervention.Event_Trigger = camp.get_send_trigger( Event_Trigger, old=old_adhoc_trigger_style )
intervention.Node_Selection_Type = Node_Selection_Type
if Max_Distance_To_Other_Nodes_Km != -1:
intervention.Max_Distance_To_Other_Nodes_Km = Max_Distance_To_Other_Nodes_Km
intervention.Include_My_Node = Include_My_Node
return intervention
cached_mid = None
[docs]def MultiInterventionDistributor(
camp,
Intervention_List
):
"""
Wrapper function to create and return a MultiInterventionDistributor intervention.
Args:
camp: emod_api.campaign object with schema_path set.
Intervention_List: List of 1 or more valid intervention dictionaries to be
distributed together.
Returns:
ReadOnlyDict: Schema-based smart dictionary representing a new
MultiInterventionDistributor intervention ready to be added to a campaign.
"""
global schema_path
schema_path = ( camp.schema_path if camp is not None else schema_path )
if Intervention_List is None or type(Intervention_List) is not list or len( Intervention_List ) == 0:
raise ValueError( "Intervention_List is empty or None or not a list." )
global cached_mid
if cached_mid is None:
cached_mid = s2c.get_class_with_defaults( "MultiInterventionDistributor", schema_path )
intervention = copy.deepcopy( cached_mid )
intervention.Intervention_List = copy.deepcopy( Intervention_List )
if "Intervention_Name" in Intervention_List[0]:
intervention.Intervention_Name = Intervention_List[0].Intervention_Name
else:
intervention.Intervention_Name = "MultiInterventionDistributor"
return intervention
[docs]def DelayedIntervention(
camp,
Configs,
Delay_Dict=None
):
"""
Wrapper function to create and return a DelayedIntervention intervention.
Args:
camp: emod_api.campaign object with schema_path set.
Config: Valid intervention config.
Delay_Dict: Dictionary of 1 or 2 params that are the literal Delay_Distribution
parameters, but without the distribution, which is inferred. E.g.,
{ "Delay_Period_Exponential": 5 }
Returns:
ReadOnlyDict: Schema-based smart dictionary representing a new
DelayedIntervention intervention ready to be added to a campaign.
"""
global schema_path
schema_path = ( camp.schema_path if camp is not None else schema_path )
if Configs is None or type( Configs ) is not list or len( Configs ) == 0:
raise ValueError( "Configs is empty or None or not a list." )
intervention = s2c.get_class_with_defaults( "DelayedIntervention", schema_path )
intervention.Actual_IndividualIntervention_Configs = copy.deepcopy( Configs )
if Delay_Dict is None:
Delay_Dict = { "Delay_Period_Constant": 0 }
for param in Delay_Dict:
setattr( intervention, param, Delay_Dict[ param ] )
return intervention
[docs]def HSB(
camp,
Event_Or_Config="Event",
Config=None,
Event="NoTrigger",
Tendency=1.0,
Single_Use=True,
Name="HSB"
):
"""
Wrapper function to create and return a HealthSeekingBehaviour intervention.
Args:
camp: emod_api.campaign object with schema_path set.
Event_Or_Config: "Event" or "Config".
Config: Complete, valid intervention configuration to be distributed.
Event: Event/Trigger/Signal to be broadcast, alternative to an intervention.
Tendency: Daily probability of 'seeking care' aka distributing payload intervention.
Single_Use: One-and-done, or continuous?
Name: Intervention Name. Useful if you want to provide uniqueness and not worry about
duplicate intervention management.
Returns:
ReadOnlyDict: Schema-based smart dictionary representing a new
HSB intervention ready to be added to a campaign.
"""
global schema_path
schema_path = ( camp.schema_path if camp is not None else schema_path )
intervention = s2c.get_class_with_defaults( "SimpleHealthSeekingBehavior", schema_path )
intervention.Event_Or_Config = Event_Or_Config
if intervention.Event_Or_Config == "Event":
intervention.Actual_IndividualIntervention_Event = camp.get_send_trigger( Event, old=old_adhoc_trigger_style )
else:
if Config is None:
raise ValueError( "You specified 'Config' but no actual Config." )
intervention.Actual_IndividualIntervention_Config = Config
intervention.Tendency = Tendency
intervention.Intervention_Name = Name
intervention.Single_Use = Single_Use
return intervention
[docs]def NLHTI(
camp,
Triggers,
Interventions,
Property_Restrictions=None,
Demographic_Coverage=1.0,
Target_Age_Min=0,
Target_Age_Max=_MAX_AGE,
Target_Gender="All",
Target_Residents_Only=False,
Duration=-1,
Blackout_Event_Trigger=None,
Blackout_Period=None,
Blackout_On_First_Occurrence=None,
Disqualifying_Properties=None,
):
"""
Wrapper function to create and return a NodeLevelHealthTriggeredIntervention intervention.
Args:
camp: emod_api.campaign object with schema_path set.
Triggers: List of Triggers/Events/Signals
Interventions: List of interventions to distrbute when signal is heard.
Property_Restrictions: Individual Properties that an agent must have to qualify for intervention.
Demographic_Coverage: Percentage of individuals to receive intervention.
Target_Age_Min: Minimum age (in years).
Target_Age_Max: Maximum age (in years).
Target_Gender: All, Male, or Female.
Target_Residents_Only: Not used.
Duration: How long this listen-and-distribute should last.
Blackout_Event_Trigger: Not used.
Blackout_Period: Not used.
Blackout_On_First_Occurrence: Not used.
Disqualifying_Properties: Not used.
Returns:
ReadOnlyDict: Schema-based smart dictionary representing a new
NLHTI intervention ready to be added to a campaign.
"""
global schema_path
schema_path = ( camp.schema_path if camp is not None else schema_path )
intervention = s2c.get_class_with_defaults( "NodeLevelHealthTriggeredIV", schema_path )
intervention.Trigger_Condition_List = [ camp.get_recv_trigger( trigger, old=old_adhoc_trigger_style ) for trigger in Triggers ]
if Target_Age_Min > 0 or Target_Age_Max < _MAX_AGE:
intervention.Target_Age_Min = Target_Age_Min
intervention.Target_Age_Max = Target_Age_Max
if Target_Gender != "All":
intervention.Target_Gender = Target_Gender
intervention.Target_Demographic = "ExplicitAgeRangesAndGender"
intervention.Target_Residents_Only = Target_Residents_Only
intervention.Demographic_Coverage = Demographic_Coverage
intervention.Duration = Duration
intervention.Blackout_Event_Trigger = Blackout_Event_Trigger
intervention.Blackout_Period = Blackout_Period
intervention.Blackout_On_First_Occurrence = Blackout_On_First_Occurrence
prs = utils._convert_prs( Property_Restrictions )
if len(prs)>0 and type(prs[0]) is dict:
intervention.Property_Restrictions_Within_Node = prs
intervention.pop( "Property_Restrictions" )
else:
intervention.Property_Restrictions = prs
if "Property_Restrictions_Within_Node" in intervention:
intervention.pop( "Property_Restrictions_Within_Node" )
if len(Interventions) == 1:
intervention.Actual_IndividualIntervention_Config = Interventions[0]
else:
intervention.Actual_IndividualIntervention_Config = MultiInterventionDistributor( camp, Interventions )
return intervention
[docs]def PropertyValueChanger(
camp,
Target_Property_Key,
Target_Property_Value,
Daily_Probability=1.0,
Maximum_Duration=1,
Revert=-1,
Intervention_Name="",
Event_Trigger_Distributed="",
Event_Trigger_Expired=""
):
"""
Wrapper function to create and return a PropertyValueChanger intervention.
Args:
camp: emod_api.campaign object with schema_path set.
Target_Property_Key. The key part of the new key-value pair of the IP.
Target_Property_Value. The value part of the new key-value pair of the IP.
New_Property_Value.. Optional IP key:value part to be set, common to all interventions.
Daily_Probability. The daily probability that an individual will move to the Target_Property_Value.
Maximum_Duration. The maximum amount of time individuals have to move to a new group. This timing works in conjunction with Daily_Probability.
Revert. The number of days before an individual moves back to their original group.
Intervention_Name. Optional Intervention_Name. Useful if managing a replacement policy.
Event_Trigger_Distributed. Optional broadcast trigger to be published when PVC is distributed.
Event_Trigger_Expired. Optional broadcast trigger to be published when PVC is expired.
Returns:
ReadOnlyDict: Schema-based smart dictionary representing a new
PropertyValueChanger intervention ready to be added to a campaign.
"""
global schema_path
schema_path = ( camp.schema_path if camp is not None else schema_path )
intervention = s2c.get_class_with_defaults( "PropertyValueChanger", schema_path )
intervention.Target_Property_Key = Target_Property_Key
intervention.Target_Property_Value = Target_Property_Value
intervention.Daily_Probability = Daily_Probability
intervention.Maximum_Duration=Maximum_Duration
if Revert != -1:
intervention.Revert=Revert
if len(Intervention_Name)>0:
intervention.Intervention_Name=Intervention_Name
if len(Event_Trigger_Distributed)>0:
intervention.Event_Trigger_Distributed=Event_Trigger_Distributed
if len(Event_Trigger_Expired)>0:
intervention.Event_Trigger_Expired=Event_Trigger_Expired
return intervention
[docs]def ScheduledCampaignEvent(
camp,
Start_Day: int,
Node_Ids = None,
Nodeset_Config = None,
Number_Repetitions: int = 1,
Timesteps_Between_Repetitions: int = -1,
Event_Name: str = "Scheduled_Campaign_Event", # not set, doesn't exist in schema
Property_Restrictions = None,
Demographic_Coverage:float = 1.0,
Target_Age_Min=0,
Target_Age_Max=_MAX_AGE,
Target_Gender:str = "All",
Target_Residents_Only:bool = False,
Intervention_List=None):
"""
Wrapper function to create and return a ScheduledCampaignEvent intervention.
The alternative to a ScheduledCampaignEvent is a TriggeredCampaignEvent.
Args:
camp: emod_api.campaign object with schema_path set.
Start_Day: When to start.
Event_Name: Name for overall campaign event, of no functional meaning. Not in schema and not yet used.
Node_Ids: Nodes to target with this intervenion
Nodeset_Config: Nodes to target with this intervenion, return from utils.do_nodes().
.. deprecated:: 2.x
Use parameter Node_Ids instead
Property_Restrictions: Individual Properties a person must have to receive the intervention(s).
Number_Repetitions: N/A
Timesteps_Between_Repetitions: N/A
Demographic_Coverage: Percentage of individuals to receive intervention.
Target_Age_Min: Minimum age (in years).
Target_Age_Max: Maximum age (in years).
Target_Gender: All, Male, or Female.
Intervention_List: List of 1 or more valid intervention dictionaries to be
distributed together.
Returns:
ReadOnlyDict: Schema-based smart dictionary representing a new
ScheduledCampaignEvent intervention ready to be added to a campaign.
"""
global schema_path
schema_path = ( camp.schema_path if camp is not None else schema_path )
# Not checking Intervention_list because MultiInterventionDistributor checks
event = s2c.get_class_with_defaults( "CampaignEvent", schema_path )
global cached_sec
if cached_sec is None:
cached_sec = s2c.get_class_with_defaults( "StandardEventCoordinator", schema_path )
coordinator = copy.deepcopy( cached_sec )
coordinator.Demographic_Coverage = Demographic_Coverage
coordinator.Number_Repetitions=Number_Repetitions
coordinator.Timesteps_Between_Repetitions=Timesteps_Between_Repetitions
# Second, hook them up
event.Event_Coordinator_Config = coordinator
event.Start_Day = float(Start_Day)
if Nodeset_Config is not None and Node_Ids is not None:
raise AssertionError("Node_Ids and Nodeset_Config are set. Please set only one.")
if Nodeset_Config is not None:
event.Nodeset_Config = Nodeset_Config
else:
# The real default is an empty list
if Node_Ids is None:
Node_Ids = []
node_ids = Node_Ids
event.Nodeset_Config = utils.do_nodes(camp.schema_path, node_ids)
coordinator.Intervention_Config = MultiInterventionDistributor( camp, Intervention_List )
prs = utils._convert_prs( Property_Restrictions )
if len(prs)>0 and type(prs[0]) is dict:
coordinator.Property_Restrictions_Within_Node = prs
coordinator.pop( "Property_Restrictions" )
else:
coordinator.Property_Restrictions = prs
coordinator.pop( "Property_Restrictions_Within_Node" )
coordinator.Demographic_Coverage = Demographic_Coverage
if Target_Age_Min > 0 or Target_Age_Max < _MAX_AGE:
coordinator.Target_Age_Min = Target_Age_Min
if Target_Age_Max is not None and Target_Age_Max < _MAX_AGE: # overwrite default from schema
coordinator.Target_Age_Max = Target_Age_Max
if Target_Gender != "All":
coordinator.Target_Gender = Target_Gender
coordinator.Target_Demographic = "ExplicitAgeRangesAndGender"
coordinator.Target_Residents_Only = Target_Residents_Only
return event
[docs]def TriggeredCampaignEvent(
camp,
Start_Day: int,
Event_Name: str,
Triggers: List[str],
Intervention_List: List[dict],
Node_Ids=None,
Nodeset_Config=None,
Node_Property_Restrictions=None,
Property_Restrictions=None,
Number_Repetitions: int = 1,
Timesteps_Between_Repetitions: int = -1,
Demographic_Coverage: float=1.0,
Target_Age_Min=0,
Target_Age_Max=_MAX_AGE,
Target_Gender: str ="All",
Target_Residents_Only=False,
Duration=-1,
Blackout_Event_Trigger: str=None,
Blackout_Period=0,
Blackout_On_First_Occurrence=0,
Disqualifying_Properties=None,
Delay=None
):
"""
Wrapper function to create and return a TriggeredCampaignEvent intervention.
The alternative to a TriggeredCampaignEvent is a ScheduledCampaignEvent.
Args:
camp: emod_api.campaign object with schema_path set.
Start_Day: When to start.
Event_Name: Name for overall campaign event, of no functional meaning. Not in schema and not yet used.
Node_Ids: Nodes to target with this intervenion
Nodeset_Config: Nodes to target with this intervenion, return from utils.do_nodes().
.. deprecated:: 2.x
Use parameter Node_Ids instead
Triggers: List of triggers/events/signals to listen to in order to trigger distribution.
Intervention_List: List of 1 or more valid intervention dictionaries to be
distributed together.
Node_Property_Restrictions: N/A.
Property_Restrictions: Individual Properties a person must have to receive the intervention(s).
Demographic_Coverage: Percentage of individuals to receive intervention.
Target_Age_Min: Minimum age (in years).
Target_Age_Max: Maximum age (in years).
Target_Gender: All, Male, or Female.
Target_Residents_Only: TBD.
Duration: How long this listen-and-distribute should last.
Blackout_Event_Trigger: Not used.
Blackout_Period: Not used.
Blackout_On_First_Occurrence: Not used.
Disqualifying_Properties: Not used.
delay: Optional delay between trigger and actual distribution.
Returns:
ReadOnlyDict: Schema-based smart dictionary representing a new
TriggeredCampaignEvent intervention ready to be added to a campaign.
"""
global schema_path
schema_path = ( camp.schema_path if camp is not None else schema_path )
if Node_Property_Restrictions is None:
Node_Property_Restrictions=[]
if Property_Restrictions is None:
Property_Restrictions=[]
global cached_ce
if cached_ce is None:
cached_ce = s2c.get_class_with_defaults( "CampaignEvent", schema_path )
event = copy.deepcopy(cached_ce)
event.Start_Day = float(Start_Day)
if Nodeset_Config is not None and Node_Ids is not None:
raise AssertionError("Node_Ids and Nodeset_Config are set. Please use only one.")
if Nodeset_Config is not None:
event.Nodeset_Config = Nodeset_Config
else:
# The real default is an empty list.
if Node_Ids is None:
Node_Ids = []
node_ids = Node_Ids
event.Nodeset_Config = utils.do_nodes(camp.schema_path, node_ids)
global cached_sec
if cached_sec is None:
cached_sec = s2c.get_class_with_defaults( "StandardEventCoordinator", schema_path )
coordinator = copy.deepcopy(cached_sec)
coordinator.Number_Repetitions = Number_Repetitions
coordinator.Timesteps_Between_Repetitions = Timesteps_Between_Repetitions
if Delay is not None:
Intervention_List = [ DelayedIntervention( camp, Intervention_List, Delay_Dict = { "Delay_Period_Constant": Delay } ) ]
intervention = NLHTI(
camp,
Triggers,
Intervention_List,
Duration=Duration,
Property_Restrictions = Property_Restrictions,
Demographic_Coverage = Demographic_Coverage,
Target_Age_Min = Target_Age_Min,
Target_Age_Max = Target_Age_Max,
Target_Gender = Target_Gender,
Target_Residents_Only = Target_Residents_Only,
Blackout_Event_Trigger=Blackout_Event_Trigger,
Blackout_Period=Blackout_Period,
Blackout_On_First_Occurrence=Blackout_On_First_Occurrence
)
event.Event_Coordinator_Config = coordinator
coordinator.Intervention_Config = intervention
return event
[docs]def StandardDiagnostic(
camp,
Base_Sensitivity: float=1.0,
Base_Specificity: float=1.0,
Days_To_Diagnosis: float=0.0,
Event_Trigger_Distributed: str = None,
Event_Trigger_Expired: str = None,
Positive_Diagnosis_Intervention = None,
Positive_Diagnosis_Event: str = "PositiveResult",
Negative_Diagnosis_Intervention = None,
Negative_Diagnosis_Event: str = "NegativeResult",
Treatment_Fraction: float=1.0
):
"""
Wrapper function to create and return a StandardDiagnostic intervention.
Args:
camp: emod_api.campaign object with schema_path set.
Base_Sensitivity: base sensitivity [0..1]
Base_Specificity: base specificity [0..1]
Days_To_Diagnosis: days to diagnosis
Event_Trigger_Distributed: A trigger that is fired when intervention was distributed
Event_Trigger_Expired: A trigger that is fired when intervention has expired
Positive_Diagnosis_Intervention: Intervention that is distributed in case of a positive diagnosis. If set, no events may be configured.
Positive_Diagnosis_Event: A trigger that is fired in case of a positive diagnosis
Negative_Diagnosis_Intervention: Intervention that is distributed in case of a Negative diagnosis. If set, no events may be configured.
Not used outside of Malaria-Ongoing yet.
Negative_Diagnosis_Event: A trigger that is fired in case of a Negative diagnosis. Not used outside of Malaria-Ongoing yet.
Treatment_Fraction: treatment fraction [0..1]
Returns:
ReadOnlyDict: Schema-based smart dictionary representing a new
MultiInterventionDistributor intervention ready to be added to a campaign.
"""
global schema_path
schema_path = ( camp.schema_path if camp is not None else schema_path )
if Positive_Diagnosis_Intervention is not None and (Event_Trigger_Distributed is not None or Event_Trigger_Expired is not None):
raise Exception("Events and intervention are configured. Configuration of events and intervention are mutually exclusive.")
# First, get the objects
# Only in Malaria-Ongoing the class is called StandardDiagnostic
try:
intervention = s2c.get_class_with_defaults("StandardDiagnostic", schema_path)
if Positive_Diagnosis_Intervention is None:
intervention.Positive_Diagnosis_Config = BroadcastEvent( camp, Positive_Diagnosis_Event )
else:
intervention.Positive_Diagnosis_Config = Positive_Diagnosis_Intervention
if Negative_Diagnosis_Intervention is None:
intervention.Negative_Diagnosis_Config = BroadcastEvent( camp, Negative_Diagnosis_Event )
else:
intervention.Negative_Diagnosis_Config = Negative_Diagnosis_Intervention
except ValueError:
# Non Malaria, use SimpleDiagnostic
intervention = s2c.get_class_with_defaults("SimpleDiagnostic", schema_path)
if Positive_Diagnosis_Intervention is None:
intervention.Positive_Diagnosis_Config = BroadcastEvent( camp, Positive_Diagnosis_Event )
# The line below works fine instead of the line above but it just makes testing harder.
#intervention.Positive_Diagnosis_Event = camp.get_send_trigger( Positive_Diagnosis_Event, old=old_adhoc_trigger_style )
intervention.Event_Trigger_Distributed = Event_Trigger_Distributed
intervention.Event_Trigger_Expired = Event_Trigger_Expired
else:
intervention.Positive_Diagnosis_Config = Positive_Diagnosis_Intervention
intervention.Base_Sensitivity = Base_Sensitivity
intervention.Base_Specificity = Base_Specificity
intervention.Days_To_Diagnosis = Days_To_Diagnosis
intervention.Treatment_Fraction = Treatment_Fraction
return intervention
[docs]def triggered_campaign_delay_event( camp, start_day, trigger, delay, intervention, ip_targeting=[], coverage=1.0 ):
"""
Create and return a campaign event that responds to a trigger after a delay with an intervention.
Args:
camp: emod_api.campaign object with schema_path set.
start_day: When to start.
delay: Dictionary of 1 or 2 params that are the literal Delay_Distribution parameters,
but without the distribution, which is inferred. E.g., { "Delay_Period_Exponential": 5 }.
trigger: E.g., "NewInfection".
intervention: List of 1 or more valid intervention dictionaries to be distributed together.
ip_targeting: Optional Individual Properties required for someone to receive the intervntion(s).
Returns:
Campaign event.
"""
delay_iv = DelayedIntervention( camp, Configs=[intervention], Delay_Dict = delay )
event = TriggeredCampaignEvent( camp, Start_Day=start_day, Event_Name="triggered_delayed_intervention", Triggers=[ camp.get_recv_trigger( trigger, old=True ) ], Intervention_List=[delay_iv], Property_Restrictions=ip_targeting, Demographic_Coverage=coverage )
return event
#
# The following function is intended to replace triggered_campaign_delay_event.
# There's a bit of a structure here:
# change_individual_property() ->
# change_individual_property_scheduled() ->
# ScheduledCampaignEvent()
# OR
# change_individual_property_triggered ->
# triggered_campaign_event_with_optional_delay
# TriggeredCampaignEvent()
[docs]def triggered_campaign_event_with_optional_delay( camp,
start_day,
triggers,
intervention,
delay=None,
duration=-1,
ip_targeting=None,
coverage=1.0,
target_age_min=0,
target_age_max=_MAX_AGE,
target_sex="All",
target_residents_only=False,
blackout=True,
check_at_trigger=False
):
"""
Create and return a campaign event that responds to a trigger after a delay with an intervention.
Args:
camp: emod_api.campaign object with schema_path set.
start_day: When to start.
triggers: List of signals to listen for/trigger on. E.g., "NewInfection".
intervention: List of 1 or more valid intervention dictionaries to be distributed together.
delay: Optional dictionary of 1 or 2 params that are the literal Delay_Distribution parameters,
but without the distribution, which is inferred. E.g., { "Delay_Period_Exponential": 5 }. If omitted,
intervention is immediate.
duration: How long to listen.
ip_targeting: Optional Individual Properties required for someone to receive the intervntion(s).
coverage: Fraction of target population to reach.
target_age_min: Minimum age to target.
target_age_max: Maximum age to target.
target_sex: Optional target just "MALE" or "FEMALE" individuals.
target_residents_only: Set to True to target only the individuals who
started the simulation in this node and are still in the node.
blackout: Set to True if you don't want the triggered intervention
to be distributed to the same person more than once a day.
check_at_trigger: if triggered event is delayed, you have an
option to check individual/node's eligibility at the initial trigger
or when the event is actually distributed after delay.
Returns:
Campaign event.
"""
if delay:
# If delay is just a number, convert to dict.
delay_iv = DelayedIntervention( camp, Configs=[intervention], Delay_Dict = delay )
event = TriggeredCampaignEvent( camp, Start_Day=start_day, Event_Name="Delayed Triggered Intervention", Triggers=triggers, Intervention_List=[delay_iv], Property_Restrictions=ip_targeting, Demographic_Coverage=coverage, Target_Age_Min=target_age_min, Target_Age_Max=target_age_max, Target_Gender=target_sex, Target_Residents_Only=target_residents_only, Duration=duration )
else:
event = TriggeredCampaignEvent( camp, Start_Day=start_day, Event_Name="Immediate Triggered Intervention", Triggers=triggers, Intervention_List=[intervention], Property_Restrictions=ip_targeting, Demographic_Coverage=coverage, Target_Age_Min=target_age_min, Target_Age_Max=target_age_max, Target_Gender=target_sex, Target_Residents_Only=target_residents_only, Duration=duration )
return event
[docs]def change_individual_property_at_age( camp, new_ip_key, new_ip_value, change_age_in_days, revert_in_days, ip_targeting_key, ip_targeting_value, coverage=1.0 ):
"""
Create and return a campaign event that changes a person's Individual Properties once they turns a certain age.
e.g., change_individual_property_at_age(cb, 'ForestGoing', 'LovesForest', coverage=0.6, change_age_in_days=15*365, revert=20*365)
Args:
camp: emod_api.campaign object with schema_path set.
new_ip_key: The new IP key.
new_ip_value: The new IP value.
change_age_in_days: The age at which the individual transitions (in units of days).
revert_in_days: How many days they remain with the new property.
ip_targeting_key: The IP key a person must have to receive this.
ip_targeting_value: The IP value a person must have to receive this.
coverage: Optional fraction to limit this to a subset of the target population.
Returns:
Campaign event.
"""
iv = PropertyValueChanger( camp, Target_Property_Key=new_ip_key, Target_Property_Value=new_ip_value, Revert=revert_in_days)
# TBD: migrate this to use triggered_campaign_event_with_optional_delay
return triggered_campaign_delay_event( camp, start_day=1, trigger="Births", coverage=coverage, delay={ "Delay_Period_Constant": change_age_in_days }, intervention=iv, ip_targeting={ ip_targeting_key : ip_targeting_value } )
[docs]def change_individual_property_triggered( camp,
triggers: list,
new_ip_key: str,
new_ip_value: str,
start_day: int=0,
daily_prob: float=1,
max_duration: int=9.3228e+35,
revert_in_days:int=-1,
node_ids: list=None, # where
ip_restrictions:list=None,
coverage: float=1.0,
target_age_min: float=0,
target_age_max: float=_MAX_AGE,
target_sex: str="All",
target_residents_only: bool=False,
delay=None,
listening_duration: int=-1,
blackout: bool=True,
check_at_trigger: bool=False
):
"""
Change Individual Properties when a certain trigger is observed.
Args:
camp: The instance containing the campaign builder and accumulator.
triggers: A list of the events that will trigger the intervention.
new_ip_key: The individual property key to assign to the
individual. For example, InterventionStatus.
new_ip_value: The individual property value to assign to the
individual. For example, RecentDrug.
start_day: The day on which to start distributing the intervention
(**Start_Day** parameter).
node_ids: The list of nodes to apply this intervention to. If not provided, defaults to all nodes.
daily_prob: The daily probability that an individual's property value
will be updated (**Daily_Probability** parameter).
max_duration: The maximum amount of time individuals have to move to a new
**daily_prob**; individuals not moved to the new value by the end of
**max_duration** keep the same value.
revert_in_days: The number of days before a node reverts to its original
property value. Default of 0 means the new value is kept forever.
ip_restrictions: The IndividualProperty key:value pairs to target.
coverage: The proportion of the population that will receive the
intervention (**Demographic_Coverage** parameter).
target_age_min: Minimum age to target.
target_age_max: Maximum age to target.
target_sex: Optional target just "MALE" or "FEMALE" individuals.
target_residents_only: Set to True to target only the individuals who
started the simulation in this node and are still in the node.
delay: The number of days the campaign is delayed after being triggered.
listening_duration: The number of time steps that the
triggered campaign will be active for. Default is -1, which is
indefinitely.
blackout (advanced): Set to True if you don't want the triggered intervention
to be distributed to the same person more than once a day.
check_at_trigger (advanced): if triggered event is delayed, you have an
option to check individual/node's eligibility at the initial trigger
or when the event is actually distributed after delay.
Returns:
N/A.
"""
iv = PropertyValueChanger( camp, Target_Property_Key=new_ip_key, Target_Property_Value=new_ip_value, Daily_Probability=daily_prob, Maximum_Duration=max_duration, Revert=revert_in_days)
tce = triggered_campaign_event_with_optional_delay( camp, start_day=start_day, intervention=iv, triggers=triggers, delay=delay, ip_targeting=ip_restrictions, coverage=coverage, target_age_min=target_age_min, target_age_max=target_age_max, target_sex=target_sex, target_residents_only=target_residents_only, duration=listening_duration, blackout=blackout, check_at_trigger=check_at_trigger )
camp.add( tce )
[docs]def change_individual_property_scheduled( camp,
new_ip_key,
new_ip_value,
start_day: int=0,
number_repetitions: int = 1,
timesteps_between_reps: int = -1,
node_ids: list=None, # where
daily_prob: float=1,
max_duration: int=9.3228e+35,
revert_in_days:int=-1,
ip_restrictions:list=None,
coverage: float=1.0,
target_age_min: float=0,
target_age_max: float=_MAX_AGE,
target_sex: str="All",
target_residents_only: bool=False
):
"""
Change Individual Properties at a given time.
Args:
camp: The instance containing the campaign builder and accumulator.
new_ip_key: The individual property key to assign to the
individual. For example, InterventionStatus.
new_ip_value: The individual property value to assign to the
individual. For example, RecentDrug.
start_day: The day on which to start distributing the intervention
(**Start_Day** parameter).
node_ids: The list of nodes to apply this intervention to. If not provided, defaults to all nodes.
daily_prob: The daily probability that an individual's property value
will be updated (**Daily_Probability** parameter).
max_duration: The maximum amount of time individuals have to move to a new
**daily_prob**; individuals not moved to the new value by the end of
**max_duration** keep the same value.
revert_in_days: The number of days before an individual reverts to its original
property value. Default of -1 means the new value is kept forever.
ip_restrictions: The IndividualProperty key:value pairs to target.
coverage: The proportion of the population that will receive the
intervention (**Demographic_Coverage** parameter).
target_age_min: Minimum age to target.
target_age_max: Maximum age to target.
target_sex: Optional target just "MALE" or "FEMALE" individuals.
target_residents_only: Set to True to target only the individuals who
started the simulation in this node and are still in the node.
Returns:
N/A.
"""
iv = PropertyValueChanger( camp, Target_Property_Key=new_ip_key, Target_Property_Value=new_ip_value, Daily_Probability=daily_prob, Maximum_Duration=max_duration, Revert=revert_in_days)
sce = ScheduledCampaignEvent( camp, Intervention_List=[iv], Start_Day=start_day, Node_Ids=node_ids, Number_Repetitions=number_repetitions, Timesteps_Between_Repetitions=timesteps_between_reps, Property_Restrictions=ip_restrictions, Demographic_Coverage=coverage, Target_Age_Min=target_age_min, Target_Age_Max=target_age_max, Target_Gender=target_sex, Target_Residents_Only=target_residents_only )
camp.add( sce )
[docs]def change_individual_property(camp,
target_property_name: str, # what...
target_property_value: str,
start_day: int=0, # when
number_repetitions: int = 1,
timesteps_between_reps: int = -1,
node_ids: list=None, # where
daily_prob: float=1,
max_duration: int=9.3228e+35,
revert: int=-1,
coverage: float=1, # who
ip_restrictions: list=None,
target_age_min: float=0,
target_age_max: float=_MAX_AGE,
target_sex: str="All",
target_residents_only: bool=False,
trigger_condition_list: list=None, # the why...
triggered_campaign_delay: int=0, # (well, kind of when again)
listening_duration: int=-1,
blackout_flag: bool=True, # advanced trigger stuff
check_eligibility_at_trigger: bool=False):
"""
Add an intervention that changes the individual property value to another on a
particular day OR after a triggering event using the
**PropertyValueChanger** class. Deprecated. Prefer change_individual_property_scheduled
or change_individual_property_triggered depending on the use case.
Args:
camp: emod_api.campaign object with schema_path set.
target_property_name: The individual property key to assign to the
individual. For example, Risk.
target_property_value: The individual property value to assign to the
individual. For example, High.
start_day: The day on which to start distributing the intervention.
number_repetitions: Optional repeater value. Does not work with triggers.
timesteps_between_reps: Gap between repetitions, optional. Does not work with triggers.
node_ids: The list of nodes to apply this intervention to. Defaults to all.
daily_prob: The daily probability that an individual's property value
will be updated (**Daily_Probability** parameter).
max_duration: The number of days to continue the intervention after
**start_day**.
revert: The number of days before an individual reverts to its original
property value. Default of -1 means the new value is kept forever.
coverage: The proportion of the population that will receive the
intervention (**Demographic_Coverage** parameter).
ip_restrictions: The IndividualProperty key:value pairs to target.
Usually this will be the same key but different from the
target_property_xxx entries.
target_residents_only: Set to True to target only the individuals who
started the simulation in this node and are still in the node.
target_age_min: Optional minimum age, defaults to 0.
target_age_max: Optional maximum age, defaults to inf.
target_sex: Optional target sex, defaults to both.
triggered_campaign_delay: The number of days the campaign is delayed
after being triggered.
trigger_condition_list: A list of the events that will
trigger the intervention. If included, **start_day** is the day
when monitoring for triggers begins.
listening_duration: The number of time steps that the
triggered campaign will be active for. Default is -1, which is
indefinitely.
blackout_flag: Set to True if you don't want the triggered intervention
to be distributed to the same person more than once a day.
check_eligibility_at_trigger: if triggered event is delayed, you have an
option to check individual/node's eligibility at the initial trigger
or when the event is actually distributed after delay.
Returns:
None
"""
if trigger_condition_list:
return change_individual_property_triggered( camp,
triggers=trigger_condition_list,
new_ip_key=target_property_name,
new_ip_value=target_property_value,
revert_in_days=revert,
ip_restrictions=ip_restrictions,
coverage=1.0,
target_age_min=target_age_min,
target_age_max=target_age_max,
target_sex=target_sex,
target_residents_only=target_residents_only,
delay=triggered_campaign_delay,
listening_duration=listening_duration,
blackout=blackout_flag,
check_at_trigger=check_eligibility_at_trigger)
else:
return change_individual_property_scheduled( camp,
new_ip_key=target_property_name,
new_ip_value=target_property_value,
revert_in_days=revert,
ip_restrictions=ip_restrictions,
coverage=1.0,
target_residents_only=target_residents_only,
target_age_min=target_age_min,
target_age_max=target_age_max,
target_sex=target_sex
)