Source code for emodpy_hiv.interventions.cascade_helpers

from typing import List

import emod_api.campaign as camp
import emodpy_hiv.interventions.outbreak as ob 
import emodpy_hiv.interventions.reftracker as rt
import emod_api.interventions.common as comm
import emod_api.interventions.utils as utils
from emodpy_hiv import *

[docs]def reset( camp ):
    """
        Utility function to clear out the campaign object, mostly useful to test. This function is not auto-imported
        with the module like the rest of the functions in this submodule. Caller has to explicitly import 
        emodpy_hiv.interventions.cascade_helpers.
        Args:
            camp: emod_api.campaign object.
        Returns:
            None.
    """
    del( camp.campaign_dict["Events"][:] )

[docs]def seed_infection( camp, timestep, coverage ):
    """
        Seed an infection by time and %age of population infected.
    """
    event = ob.new_intervention( timestep=timestep, camp=camp, coverage=coverage )
    # have to be careful with having first in a wrapped function. This removes all other events.
    camp.add( event )


# TODO: need to decide: use kwargs or custom pass-through params that go to TriggerdCampaignEvent (e.g. target_age_max)
[docs]def triggered_event_common(camp, in_trigger, out_iv, coverage=1.0, target_sex="All",
                           property_restrictions: List[str] = None, event_name="", node_ids: List[int] = None,
                           start_day: int = 1, target_age_min: float = 0, target_age_max: float = 365*200, delay: float = None,
                           duration: float = -1.0):
    """
        Parameterized utility function used by rest of functions in this submodule; listens for a trigger (signal)
        and distributes an intervention (or list thereof) as a result, based on coverage and targeting. 

        Args: 
            camp: The central campaign accumulator, from emod_api.campaign.  
            in_trigger: The trigger aka signal for the event. E.g., "NewInfection"
            out_iv: The output intervention of the event.  
            coverage (float): The coverage of the event (default is 1.0).  
            target_sex (str): The target sex for the event (default is "All").  
            property_restrictions (list[str]): A list of IP restrictions in "Risk:HIGH" format.
                None means no restrictions.
            event_name (str): The name of the event (default is an empty string).
            node_ids: (list[int]) A list of node ids to apply the event to. None means all nodes.
            start_day: (int) The model simulation time to start the event.
            duration: How long this listen-and-distribute should last.
        Returns:
            New campaign event.
    """

    out_ivs = out_iv if isinstance(out_iv, list) else [out_iv]
    in_triggers = in_trigger if isinstance(in_trigger, list) else [in_trigger]
    event = comm.TriggeredCampaignEvent(camp, Start_Day=start_day, Event_Name=event_name,
                                        Nodeset_Config=utils.do_nodes(camp.schema_path, node_ids=node_ids),
                                        Triggers=in_triggers, Intervention_List=out_ivs, Demographic_Coverage=coverage,
                                        Target_Gender=target_sex, Property_Restrictions=property_restrictions,
                                        Target_Age_Min=target_age_min, Target_Age_Max=target_age_max, Delay=delay,
                                        Duration=duration)
    # TODO: ask, should we ALWAYS pop these targeting configs? I think not, once they are implemented...
    event.Event_Coordinator_Config.pop("Targeting_Config", None)
    event.Event_Coordinator_Config.Intervention_Config.pop("Targeting_Config", None)
    event['Event_Name'] = event_name  # TODO: This should be in emod-api comm.TriggeredCampaignEvent, right?
    camp.add(event)


# An alias for triggered_event_common.
add_triggered_event = triggered_event_common


[docs]def add_scheduled_event(camp, out_iv, coverage=1.0, target_sex="All",
                        property_restrictions: List[str] = None, event_name="", node_ids: List[int] = None,
                        start_day: int = 1):
    out_ivs = out_iv if isinstance(out_iv, list) else [out_iv]
    event = comm.ScheduledCampaignEvent(camp, Start_Day=start_day, Event_Name=event_name,
                                        Nodeset_Config=utils.do_nodes(camp.schema_path, node_ids=node_ids),
                                        Intervention_List=out_ivs, Demographic_Coverage=coverage,
                                        Target_Gender=target_sex, Property_Restrictions=property_restrictions)
    # TODO: ck4, needed? (following two lines)
    event.Event_Coordinator_Config.pop("Targeting_Config", None)
    event.Event_Coordinator_Config.Intervention_Config.pop("Targeting_Config", None)
    event['Event_Name'] = event_name  # TODO: This should be in emod-api comm.TriggeredCampaignEvent, right?
    camp.add(event)


[docs]def add_reference_tracked_event(camp,
                                out_iv,
                                time_value_map: dict,
                                event_name: str = '',
                                target_min_age: float = 0,
                                target_max_age: float = 200,
                                target_sex: str = 'All',
                                target_disease_state: str = None,
                                update_period: float = 30.4166666666667,
                                start_day: float = 1960.5,
                                node_ids: List[int] = None):
    # out_ivs = out_iv if isinstance(out_iv, list) else [out_iv]
    event = rt.DistributeIVByRefTrack(camp=camp,
                                      Start_Day=start_day,
                                      Intervention=out_iv,
                                      TVMap=time_value_map,
                                      node_ids=node_ids,
                                      Target_Age_Min=target_min_age,
                                      Target_Age_Max=target_max_age,
                                      Target_Gender=target_sex,
                                      Update_Period=update_period,
                                      Event_Name=event_name,
                                      Target_Disease_State=target_disease_state)
    # event.Event_Name = event_name
    event['Event_Name'] = event_name  # TODO: This should be in emodpy-hiv rt.DistributeIVByRefTrack, right?
    camp.add(event)


[docs]def add_choice( camp, sympto_signal="NewlySymptomatic", get_tested_signal="GetTested" ):
    """
        Listen for a trigger. Defaults to 'NewlySymptomatic'. Then toss a fair coin; in case of heads,
        Distribute diagnostic (and publish signal -- default is 'GetTested'; tails maybe it's just a cold ('Ignore').

        Args:
            camp: The central campaign accumulator, from emod_api.campaign.
            sympto_signal (str): The signal or trigger we are listening for. Doesn't have to be anything to do with 
        symptomaticity but defaults to "NewlySymptomatic".  
            get_tested_signal (str): The signal/trigger published if coin toss is heads and person gets tested.  
    """
    actual_iv = randomchoice.new_diagnostic( camp, choices={ get_tested_signal: 0.5, "Ignore": 0.5 } )
    triggered_event_common( camp, in_trigger=sympto_signal, out_iv=actual_iv, event_name="Decide_On_Testing" )
 
[docs]def add_test( camp, get_tested_signal="GetTested" ):
    """
        Listen for get_tested_signal signal (defaults to "GetTested"). Then give HIV RapidDiagnostic test after a delay of 30 days.
        Publish 'HIVPositiveTest' on +ve or 'HIVNegativeTest' on -ve.
    """
    actual_iv = rapiddiag.new_diagnostic( camp, Positive_Event="HIVPositiveTest", Negative_Event="HIVNegativeTest" )
    delayed_iv = comm.DelayedIntervention( camp, [ actual_iv ], Delay_Dict={ "Delay_Period_Constant": 30 } )
    triggered_event_common( camp, get_tested_signal, delayed_iv, event_name="Test" )

[docs]def trigger_art_from_pos_test( camp, input_signal="HIVPositiveTest", output_signal="StartTreatment", lag_time=30 ):
    """
        Listen for HIVPositiveTest trigger. Then Trigger ART. Note that Trigger ART isn't same as starting it.
    """
    actual_iv = comm.BroadcastEvent( camp, output_signal )
    delayed_iv = comm.DelayedIntervention( camp, [ actual_iv ], Delay_Dict={ "Delay_Period_Constant": lag_time } )
    triggered_event_common( camp, input_signal, delayed_iv, event_name="NeedTreatment" )

[docs]def add_art_from_trigger( camp, signal="StartTreatment" ):
    """
        Actually distribute ART if a StartTreatment signal/trigger is observed. Broadcast a StartedART signal synchronously.
    """
    actual_iv = art.new_intervention(camp)
    triggered_event_common( camp, signal, [ actual_iv ], event_name="ART on trigger" )

[docs]def trigger_art( camp, timestep, coverage, trigger="StartTreatment" ):
    """
        Schedule a broadcast of StartTreatment (or equivalent), not based on any observed signals.
    """
    trigger_event = comm.ScheduledCampaignEvent( camp, Start_Day=timestep, Demographic_Coverage=coverage, Event_Name="ART trigger", Nodeset_Config=utils.do_nodes(camp.schema_path, node_ids=[]), Intervention_List=[ comm.BroadcastEvent( camp, trigger ) ] )
    trigger_event.Event_Coordinator_Config.pop( "Targeting_Config" )
    camp.add( trigger_event )