Source code for emodpy_malaria.interventions.diag_survey

"""
This module contains functionality for diagnostic survey interventions.
"""
import random
import emod_api.interventions.utils as utils
from emod_api import schema_to_class as s2c
from emod_api.interventions.common import *
from emodpy_malaria.interventions.common import _malaria_diagnostic, add_campaign_event



[docs]
def add_diagnostic_survey(
        campaign,
        coverage: float = 1,
        repetitions: int = 1,
        tsteps_btwn_repetitions: int = 365,
        target: object = 'Everyone',
        start_day: int = 1,
        diagnostic_type: str = 'BLOOD_SMEAR_PARASITES',
        diagnostic_threshold: float = 40,
        measurement_sensitivity: float = 0.1,
        event_name: str = "Diagnostic Survey",
        node_ids: list = None,
        positive_diagnosis_configs: list = None,
        negative_diagnosis_configs: list = None,
        received_test_event: str = 'Received_Test',
        ind_property_restrictions: list = None,
        disqualifying_properties: list = None,
        trigger_condition_list: list = None,
        listening_duration: int = -1,
        triggered_campaign_delay: int = 0,
        check_eligibility_at_trigger: bool = False,
        expire_recent_drugs: any = None):
    """
        Add an intervention to create either a scheduled or a triggered event to the
        campaign using the
        :py:class:`~emodpy_malaria.interventions.common.MalariaDiagnostic` class, an
        individual-level class, to test individuals. Upon positive or negative
        diagnosis, the list of events to occur (as defined in
        **positive_diagnosis_configs** or **negative_diagnosis_configs**) is then executed.
        These events can trigger other listening interventions.

        Args:
            camp: The :py:obj:`emod_api:emod_api.campaign` object for building,
                modifying, and writing campaign configuration files.

            coverage: The probability an individual receives the diagnostic.

            repetitions: Number of repetitions of the survey intervention.

            tsteps_btwn_repetitions: Timesteps between repetitions.

            target: A dictionary targeting an age range and gender of
                individuals for treatment. In the format
                ``{"agemin": x, "agemax": y, "gender": z}``. Default is Everyone.

            start_day: The day of the simulation on which the intervention is created.
                If triggered, runs on trigger, not on **start_day**.

            diagnostic_type: Type of malaria diagnostic used. Default is
                **BLOOD_SMEAR_PARASITES**. Available options are:

                * BLOOD_SMEAR_PARASITES
                * BLOOD_SMEAR_GAMETOCYTES
                * PCR_PARASITES
                * PCR_GAMETOCYTES
                * PF_HRP2
                * TRUE_INFECTION_STATUS
                * TRUE_PARASITE_DENSITY
                * FEVER

            diagnostic_threshold: The diagnostic detection threshold based on
                **diagnostic_type**:

                TRUE_INFECTION_STATUS

                BLOOD_SMEAR_PARASITES
                    In parasites/microliter, use measurement = float( 1.0 / measurementSensitivity *
                    Poisson(measurementSensitivity * true_parasite_density)).
                BLOOD_SMEAR_GAMETOCYTES
                    In gametocytes/microliter, use measurement = float( 1.0 / measurementSensitivity *
                    Poisson(measurementSensitivity * true_gametocyte_density)).
                PCR_PARASITES and PCR_GAMETOCYTES
                    Use the true values and an algorithm based on the paper
                    `Improving statistical inference on pathogen densities estimated by
                    quantitative molecular methods : malaria gametocytaemia as a case study
                    <https://doi.org/10.1186/s12859-014-0402-2>`_.
                PF_HRP2
                    Add a new method to get the PfHRP2 value and check against
                    the threshold.
                TRUE_PARASITE_DENSITY
                    Check the true parasite density against the threshold.
                FEVER
                    Check the person's fever against the threshold.

            measurement_sensitivity: Setting for **Measurement_Sensitivity** in
                :py:class:`~emodpy_malaria.interventions.common.MalariaDiagnostic`.

            event_name: Description of the event.

            node_ids: The list of nodes to apply this intervention to (**Node_List**
                parameter). If not provided, set value of NodeSetAll.

            positive_diagnosis_configs: List of events to happen to an individual
                who receives a positive result from test.

            negative_diagnosis_configs: List of events to happen to individual who
                receives a negative result from test.

            received_test_event: String for individuals to broadcast upon receiving
                diagnostic.

            ind_property_restrictions: List of IndividualProperty key:value pairs that
                individuals must have to receive the diagnostic intervention.
                For example, ``[{"IndividualProperty1":"PropertyValue1"},
                {"IndividualProperty2":"PropertyValue2"}]``. Default is no
                restrictions.

            disqualifying_properties: List of IndividualProperty key:value pairs that
                cause an intervention to be aborted. For example,
                ``[{"IndividualProperty1":"PropertyValue1"},
                {"IndividualProperty2":"PropertyValue2"}]``.

            trigger_condition_list: List of events that will trigger a diagnostic survey.

            listening_duration: Duration after start day to stop listening for events, as
                specified in **trigger_condition_list**. Default is -1, non-stop
                listening.

            triggered_campaign_delay: Delay of running the intervention after receiving
                a trigger from the **trigger_condition_list**.

            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.

            expire_recent_drugs: Adds ``[{"DrugStatus:None"}]`` to
                **Property_Restrictions_Within_Node** for positive test config, so
                only those with that property receive drugs.

        Returns:
           None
    """

    if ind_property_restrictions is None:
        ind_property_restrictions = []
    if disqualifying_properties is None:
        disqualifying_properties = []

    received_test_event = BroadcastEvent(campaign, Event_Trigger=received_test_event)

    tested_positive = BroadcastEvent(campaign, Event_Trigger="TestedPositive")
    tested_negative = BroadcastEvent(campaign, Event_Trigger="TestedNegative")
    tested_positive_tether = "TestedPositive_{}".format(random.randint(1, 100000))
    tested_negative_tether = "TestedNegative_{}".format(random.randint(1, 100000))

    intervention_cfg = _malaria_diagnostic(
        campaign,
        measurement_sensitivity=measurement_sensitivity,
        detection_threshold=diagnostic_threshold,
        diagnostic_type=diagnostic_type)

    bcast = BroadcastEvent(campaign, Event_Trigger=tested_positive_tether)
    mid = MultiInterventionDistributor(campaign, Intervention_List=[tested_positive, bcast])

    intervention_cfg.Positive_Diagnosis_Config = mid

    intervention_cfg.Negative_Diagnosis_Config = MultiInterventionDistributor(
        campaign, Intervention_List=[tested_negative, BroadcastEvent(campaign, Event_Trigger=tested_negative_tether)])

    interventions = [intervention_cfg, received_test_event]

    gender = "All"
    age_min = 0
    age_max = 9.3228e+35
    if target != "Everyone" and isinstance(target, dict):
        try:
            age_min = target["agemin"]
            age_max = target["agemax"]
            if 'gender' in target:
                gender = target["gender"]
                target = "ExplicitAgeRangesAndGender"
            else:
                target = "ExplicitAgeRanges"
        except KeyError:
            raise KeyError("Unknown target_group parameter. Please pass in 'Everyone' or a dictionary of "
                           "{'agemin' : x, 'agemax' : y, 'gender': 'Female'} to target  to individuals between x and "
                           "y years of age, and (optional) gender.\n")

    if trigger_condition_list:
        if listening_duration == -1:
            diagnosis_config_listening_duration = -1
        else:
            diagnosis_config_listening_duration = listening_duration + 1
        # if once triggered, you want the diagnostic survey to repeat or if there is a delay (or both)
        # this creates a series of broadcast events that once triggered send out "Diagnostic_Survey_Now"
        # at pre-determined intervals
        if repetitions > 1 or triggered_campaign_delay > 0:
            # create a trigger for each of the delays.
            trigger_ind_property_restrictions = []
            if check_eligibility_at_trigger:
                trigger_ind_property_restrictions = ind_property_restrictions
                ind_property_restrictions = []
            broadcast_event = "Diagnostic_Survey_Now_{}".format(random.randint(1, 100000))
            for x in range(repetitions):
                tcde = TriggeredCampaignEvent(
                    campaign,
                    Start_Day=start_day + 1,
                    Event_Name="Diag_Survey_Now",
                    Node_Ids=node_ids,
                    Triggers=trigger_condition_list,
                    Duration=listening_duration,
                    Intervention_List=[BroadcastEvent(campaign, broadcast_event)],
                    Property_Restrictions=trigger_ind_property_restrictions,
                    Delay=triggered_campaign_delay + (x * tsteps_btwn_repetitions))
                campaign.add(tcde)
                trigger_condition_list = [broadcast_event]

        survey_event = TriggeredCampaignEvent(
            campaign,
            Start_Day=start_day + 1,
            Event_Name=event_name,
            Node_Ids=node_ids,
            Triggers=trigger_condition_list,
            Target_Residents_Only=1,
            Duration=listening_duration,
            Demographic_Coverage=coverage,
            Target_Age_Min=age_min,
            Target_Age_Max=age_max,
            Target_Gender=gender,
            Property_Restrictions=ind_property_restrictions,
            Disqualifying_Properties=disqualifying_properties,
            Intervention_List=interventions)

        campaign.add(survey_event)

    else:
        diagnosis_config_listening_duration = listening_duration
        add_campaign_event(
            campaign,
            start_day=start_day + 1,
            node_ids=node_ids,
            ind_property_restrictions=ind_property_restrictions,
            repetitions=repetitions,
            timesteps_between_repetitions=tsteps_btwn_repetitions,
            demographic_coverage=coverage,
            target_age_min=age_min,
            target_age_max=age_max,
            target_gender=gender,
            individual_intervention=interventions)

    if expire_recent_drugs:
        if ind_property_restrictions:
            for property_restriction in ind_property_restrictions:
                property_restriction["DrugStatus"] = "None"
        else:
            ind_property_restrictions = [{"DrugStatus": "None"}]

    if positive_diagnosis_configs:
        tested_positive_event = TriggeredCampaignEvent(
            campaign,
            Start_Day=start_day,
            Event_Name=event_name + "Positive Result Action",
            Node_Ids=node_ids,
            Duration=diagnosis_config_listening_duration,
            Property_Restrictions=ind_property_restrictions,
            Triggers=[tested_positive_tether],
            Intervention_List=positive_diagnosis_configs
        )
        campaign.add(tested_positive_event)

    if negative_diagnosis_configs:
        tested_negative_event = TriggeredCampaignEvent(
            campaign,
            Start_Day=start_day,
            Event_Name=event_name + "Negative Result Action",
            Node_Ids=node_ids,
            Duration=diagnosis_config_listening_duration,
            Triggers=[tested_negative_tether],
            Intervention_List=negative_diagnosis_configs
        )
        campaign.add(tested_negative_event)

    return