from typing import List
from emod_api import schema_to_class as s2c
from emod_api.interventions import common, utils
def _malaria_diagnostic(
campaign,
diagnostic_type: str = "BLOOD_SMEAR_PARASITES",
measurement_sensitivity: float = 0,
detection_threshold: float = 0):
"""
Configures individual-targeted MalariaDiagnostic intervention
Args:
campaign: The :py:obj:`emod_api:emod_api.campaign` object to which the intervention
will be added.
diagnostic_type: The setting for **Diagnostic_Type** in
:doc:`emod/parameter-campaign-individual-malariadiagnostic`.
In addition to the accepted values listed there, you may also set
TRUE_INFECTION_STATUS, which calls
:doc:`emod/parameter-campaign-individual-standarddiagnostic`
instead.
measurement_sensitivity: The setting for **Measurement_Sensitivity**
in :doc:`emod/parameter-campaign-individual-malariadiagnostic`.
detection_threshold: The setting for **Detection_Threshold** in
:doc:`emod/parameter-campaign-individual-malariadiagnostic`.
Returns:
Configured individual-targeted MalariaDiagnostic intervention
"""
# Shares lots of code with Standard. Not obvious if code minimization maximizes readability.
import emod_api.interventions.common as emodapi_com
schema_path = campaign.schema_path
# First, get the objects
if diagnostic_type == "TRUE_INFECTION_STATUS":
if measurement_sensitivity or detection_threshold:
raise ValueError("MalariaDiagnostic invoked with 'TRUE_INFECTION_STATUS' and values "
"of either measurement_sensitivity or detection_threshold params (or both). "
"Those parameters are not used for TRUE_INFECTION_STATUS.")
intervention = emodapi_com.StandardDiagnostic(campaign)
else:
intervention = s2c.get_class_with_defaults("MalariaDiagnostic", schema_path)
intervention.Measurement_Sensitivity = measurement_sensitivity
intervention.Detection_Threshold = detection_threshold
intervention.Diagnostic_Type = diagnostic_type
return intervention
[docs]
def add_triggered_campaign_delay_event(campaign,
start_day: int = 1,
trigger_condition_list: list = None,
listening_duration: int = -1,
delay_period_constant: float = 0,
demographic_coverage: float = 1.0,
node_ids: list = None,
repetitions: int = 1,
timesteps_between_repetitions: int = 365,
ind_property_restrictions: list = None,
disqualifying_properties: list = None,
target_age_min: float = 0,
target_age_max: float = 125,
target_gender: str = "All",
target_residents_only: bool = False,
blackout_event_trigger: str = None,
blackout_period: float = 0,
blackout_on_first_occurrence: bool = 0,
individual_intervention: any = None):
"""
Create and add campaign event that responds to a trigger after an optional delay with an intervention.
Args:
campaign: campaign object to which the intervention will be added, and schema_path container
start_day: The day the intervention is given out.
trigger_condition_list: A list of the events that will trigger intervention distribution.
listening_duration: The number of time steps that the distributed event will monitor for triggers.
Default is -1, which is indefinitely.
delay_period_constant: Optional. Delay, in days, before the intervention is given out after a trigger
is received.
demographic_coverage: This value is the probability that each individual in the target population will
receive the intervention. It does not guarantee that the exact fraction of the target population set by
Demographic_Coverage receives the intervention.
node_ids: List of nodes to which to distribute the intervention. [] or None, indicates all nodes
will get the intervention
repetitions: The number of times an intervention is given, used with timesteps_between_repetitions. -1 means
the intervention repeats forever. Sets **Number_Repetitions**
timesteps_between_repetitions: The interval, in timesteps, between repetitions. Ignored if repetitions = 1.
Sets **Timesteps_Between_Repetitions**
ind_property_restrictions: A list of dictionaries of IndividualProperties, which are needed for the individual
to receive the intervention. Sets the **Property_Restrictions_Within_Node**
disqualifying_properties: A list of IndividualProperty key:value pairs that cause an intervention to be aborted.
Generally used to control the flow of health care access. For example, to prevent the same individual from
accessing health care via two different routes at the same time.
target_age_min: The lower end of ages targeted for an intervention, in years. Sets **Target_Age_Min**
target_age_max: The upper end of ages targeted for an intervention, in years. Sets **Target_Age_Max**
target_gender: The gender targeted for an intervention: All, Male, or Female.
target_residents_only: When set to True, the intervention is only distributed to individuals that began
the simulation in the node (i.e. those that claim the node as their residence)
blackout_event_trigger: The event to broadcast if an intervention cannot be distributed due to the
**Blackout_Period**.
blackout_period: After the initial intervention distribution, the time, in days, to wait before distributing
the intervention again. If it cannot distribute due to the blackout period, it will broadcast the
user-defined **Blackout_Event_Trigger**.
blackout_on_first_occurrence: If set to true (1), individuals will enter the blackout period after the first
occurrence of an event in the **Trigger_Condition_List***.
individual_intervention: Individual intervention or a list of individual interventions to be distributed
by this event
Returns:
Nothing
"""
if not trigger_condition_list:
raise ValueError(f"Please define trigger_condition_list.\n")
event = common.TriggeredCampaignEvent(camp=campaign,
Start_Day=start_day,
Event_Name="TriggeredEvent",
Triggers=trigger_condition_list,
Intervention_List=individual_intervention if isinstance(
individual_intervention, list) else [individual_intervention],
Node_Ids=node_ids,
Timesteps_Between_Repetitions=timesteps_between_repetitions,
Number_Repetitions=repetitions,
Target_Gender=target_gender,
Target_Age_Max=target_age_max,
Target_Age_Min=target_age_min,
Target_Residents_Only=1 if target_residents_only else 0,
Duration=listening_duration,
Demographic_Coverage=demographic_coverage,
Delay=delay_period_constant,
Disqualifying_Properties=disqualifying_properties,
Blackout_Period=blackout_period,
Blackout_Event_Trigger=blackout_event_trigger,
Blackout_On_First_Occurrence=blackout_on_first_occurrence
)
triggered_event = event.Event_Coordinator_Config.Intervention_Config
individual_restrictions = utils._convert_prs(ind_property_restrictions)
if len(individual_restrictions) > 0 and type(individual_restrictions[0]) is dict:
triggered_event["Property_Restrictions_Within_Node"] = individual_restrictions
else:
triggered_event.Property_Restrictions = individual_restrictions
campaign.add(event)
[docs]
def add_campaign_event(campaign,
start_day: int = 1,
demographic_coverage: float = 1.0,
target_num_individuals: int = None,
node_ids: list = None,
repetitions: int = 1,
timesteps_between_repetitions: int = 365,
ind_property_restrictions: list = None,
target_age_min: float = 0,
target_age_max: float = 125,
target_gender: str = "All",
target_residents_only: bool = False,
individual_intervention: any = None,
node_intervention: any = None):
"""
Adds a campaign event to the campaign with a passed in intervention.
Args:
campaign: campaign object to which the intervention will be added, and schema_path container
start_day: The day the intervention is given out.
demographic_coverage: This value is the probability that each individual in the target population will
receive the intervention. It does not guarantee that the exact fraction of the target population set by
Demographic_Coverage receives the intervention.
target_num_individuals: The exact number of people to select out of the targeted group. If this value is set,
demographic_coverage parameter is ignored
node_ids: List of nodes to which to distribute the intervention. [] or None, indicates all nodes
will get the intervention
repetitions: The number of times an intervention is given, used with timesteps_between_repetitions. -1 means
the intervention repeats forever. Sets **Number_Repetitions**
timesteps_between_repetitions: The interval, in timesteps, between repetitions. Ignored if repetitions = 1.
Sets **Timesteps_Between_Repetitions**
ind_property_restrictions: A list of dictionaries of IndividualProperties, which are needed for the individual
to receive the intervention. Sets the **Property_Restrictions_Within_Node**
target_age_min: The lower end of ages targeted for an intervention, in years. Sets **Target_Age_Min**
target_age_max: The upper end of ages targeted for an intervention, in years. Sets **Target_Age_Max**
target_gender: The gender targeted for an intervention: All, Male, or Female.
target_residents_only: When set to True, the intervention is only distributed to individuals that began
the simulation in the node (i.e. those that claim the node as their residence)
individual_intervention: Individual intervention or a list of individual interventions to be distributed
by this event
node_intervention: Node intervention or a list of node interventions to be distributed
by this event
Returns:
Nothing
"""
if individual_intervention and node_intervention:
raise ValueError(f"You cannot define both individual_intervention and node_intervention, only one.\n")
elif not individual_intervention and not node_intervention:
raise ValueError(f"Please pass in either individual_intervention or node_intervention.\n")
if individual_intervention:
event = common.ScheduledCampaignEvent(camp=campaign,
Start_Day=start_day,
Node_Ids=node_ids,
Number_Repetitions=repetitions,
Timesteps_Between_Repetitions=timesteps_between_repetitions,
Event_Name="ScheduledCampaignEvent",
Demographic_Coverage=demographic_coverage,
Property_Restrictions=ind_property_restrictions,
Target_Age_Min=target_age_min,
Target_Age_Max=target_age_max,
Target_Gender=target_gender,
Target_Residents_Only=target_residents_only,
Intervention_List=individual_intervention if isinstance(
individual_intervention,
list) else [
individual_intervention])
event.Event_Coordinator_Config.Target_Num_Individuals = target_num_individuals
campaign.add(event)
else:
schema_path = campaign.schema_path
event = s2c.get_class_with_defaults("CampaignEvent", schema_path)
event.Start_Day = start_day
event.Nodeset_Config = utils.do_nodes(schema_path, node_ids)
if isinstance(node_intervention, list):
multi_intervention_distributor = s2c.get_class_with_defaults("MultiNodeInterventionDistributor",
schema_path)
multi_intervention_distributor.Node_Intervention_List = node_intervention
intervention = multi_intervention_distributor
else:
intervention = node_intervention
# configuring the coordinator
coordinator = s2c.get_class_with_defaults("StandardEventCoordinator", schema_path)
if target_num_individuals is not None:
coordinator.Target_Num_Individuals = target_num_individuals
else:
coordinator.Demographic_Coverage = demographic_coverage
coordinator.Number_Repetitions = repetitions
coordinator.Timesteps_Between_Repetitions = timesteps_between_repetitions
coordinator.Property_Restrictions_Within_Node = ind_property_restrictions if ind_property_restrictions else []
coordinator.Property_Restrictions = [] # not using; Property_Restrictions_Within_Node are more flexible
if target_age_min > 0 or target_age_max < 125:
coordinator.Target_Age_Min = target_age_min
coordinator.Target_Age_Max = target_age_max
if target_gender != "All":
coordinator.Target_Gender = target_gender
coordinator.Target_Demographic = "ExplicitAgeRangesAndGender"
event.Event_Coordinator_Config = coordinator
coordinator.Intervention_Config = intervention
campaign.add(event)
