Source code for emod_api.interventions.node_multiplier

from .. import schema_to_class as s2c
import json
import emod_api.campaign as camp
import emod_api.interventions.utils as utils

"""
    if(camp_obj['Nodeset_Config']['Node_List'][0] == 3):
      xval = np.arange(0,365)
      yval = 1.0+0.2*np.sin(8.0*2*np.pi*xval/365)
      camp_obj['Start_Day'] = 180
      camp_obj[ECC][IC][MBD]['Times']  = xval.tolist()
      camp_obj[ECC][IC][MBD]['Values'] = yval.tolist()

    if(camp_obj['Nodeset_Config']['Node_List'][0] == 2):
      xval = np.array([ 0.0, 50.0, 100.0, 149.5, 150.0, 200.0])
      yval = np.array([ 1.0,  1.3,   1.0,   1.0,   0.7,   0.7])
      camp_obj['Start_Day'] = 10
      camp_obj[ECC][IC][MBD]['Times']  = xval.tolist()
      camp_obj[ECC][IC][MBD]['Values'] = yval.tolist()

    if(camp_obj['Nodeset_Config']['Node_List'][0] == 1):
      xval = np.arange(0,365)
      yval = 1.0-0.8*np.exp(-5.0*xval/365)
      camp_obj['Start_Day'] = 0
      camp_obj[ECC][IC][MBD]['Times']  = xval.tolist()
      camp_obj[ECC][IC][MBD]['Values'] = yval.tolist()
"""
[docs]def new_intervention( camp, new_infectivity=1.0, profile="CONST", **kwargs ):
    """
        Create new NodeInfectivityModifying intervention.

        Args:
            profile: multiplier options include:

                - CONST(ANT)
                    - new_infectivity lasts forever (or until replaced). 
                - TRAP(EZOID)
                    - rise_dur(ation)
                    - peak_dur(ation)
                    - fall_dur(ation)
                - EXP(ONENTIAL) (not implemented yet)
                    - rise duration
                    - rise rate
                - SIN(USOIDAL) (not implemented yet)
                    - period

            To do boxcar, specify 0 rise and fall durations (or omit). To do sawtooth, specify 0 peak duration (or omit).

        Returns: 
            new NodeInfectivityMult intervention dictionary.
    """
    intervention = s2c.get_class_with_defaults( "NodeInfectivityMult", camp.schema_path )
    if profile.startswith("CONST"):
        intervention.Multiplier_By_Duration.Times=[0, 365000] # I'm preferring 1000 years to FLT_MAX
        intervention.Multiplier_By_Duration.Values=[new_infectivity,new_infectivity]
    elif profile.startswith("TRAP"):
        rise_dur = 0
        if "rise_dur" in kwargs:
            rise_dur = float(kwargs["rise_dur"])
        peak_dur = 0
        if "peak_dur" in kwargs:
            peak_dur = float(kwargs["peak_dur"])
        fall_dur = 0
        if "fall_dur" in kwargs:
            fall_dur = float(kwargs["fall_dur"])
        total_durs = rise_dur + peak_dur + fall_dur
        if total_durs == 0:
            raise ValueError( "Didn't find any durations." )
        elif total_durs >365:
            raise ValueError( "Total of durations needs to be less than a year." )
        rise_dur = max(rise_dur,1)
        peak_dur = max(peak_dur,1)
        fall_dur = max(fall_dur,1)
        total_durs = rise_dur + peak_dur + fall_dur
        intervention.Multiplier_By_Duration.Times=[0, rise_dur, rise_dur+peak_dur, rise_dur+peak_dur+fall_dur]
        intervention.Multiplier_By_Duration.Values=[1.0, new_infectivity, new_infectivity, 1.0]
    else:
        raise ValueError( f"profile {profile} not supported at this time. Valid profiles are: 'CONST', 'TRAP'." )
    return intervention

[docs]def new_scheduled_event( camp, start_day=1, new_infectivity=1.0, profile="CONST", node_ids=None, recurring=True, **kwargs ):
    """
        Create new NodeInfectivityModifying intervention as scheduled campaign event.
    """
    #new_iv = new_intervention( camp=camp, new_infectivity=new_infectivity, profile=profile, kwargs=**kwargs )
    new_iv = new_intervention( camp, new_infectivity, profile, **kwargs )

    coordinator = s2c.get_class_with_defaults( "StandardEventCoordinator", camp.schema_path )
    coordinator.Intervention_Config = new_iv
    if recurring == False:
        coordinator.Number_Repetitions = 1
    else:
        coordinator.Number_Repetitions = -1
        coordinator.Timesteps_Between_Repetitions=365

    event = s2c.get_class_with_defaults( "CampaignEvent", camp.schema_path )
    event.Nodeset_Config = utils.do_nodes( camp.schema_path, node_ids )
    event.Event_Coordinator_Config = coordinator
    event.Start_Day = start_day

    return event

[docs]def new_intervention_as_file( camp, timestep, filename=None ):
    """
    Create new NodeInfectivityModifying intervention as sole scheduled campaign event inside working campaign json file.
    """
    camp.add( new_scheduled_event( camp, profile="CONSTANT", start_day=timestep, node_ids=[] ))
    if filename is None:
        filename = "node_infectivity.json"
    camp.save( filename )
    return filename