Random Trial

name: str

Name of feature in the params file. Also used to name the attribute in Agent

stats: List[str]

Random Trial collects the following stats:

• random_trial - number of agents with active random_trial
• random_trial_treated - number of active agents treated
• random_trial_treated_hiv - number of HIV+ agents treated
• random_trial_suitable - number of active agents suitable

set_stats(self, stats, time)

Update the stats dictionary passed for this agent. Called from output.get_stats for each enabled feature in the model.

The stats to be updated must be declared in the class attribute stats to make sure the dictionary has the expected keys/counter value initialized.

Parameters:

Name	Type	Description	Default
stats	Dict[str, int]	the dictionary to update with this agent's feature statistics	required
time	int	the time step of the model when the stats are set	required

Source code in titan/features/random_trial.py

def set_stats(self, stats: Dict[str, int], time: int):
    if self.active:
        stats["random_trial"] += 1
        if self.treated:
            stats["random_trial_treated"] += 1
            if self.agent.hiv.active:  # type: ignore[attr-defined]
                stats["random_trial_treated_hiv"] += 1
        if self.suitable:
            stats["random_trial_suitable"] += 1

update_pop(model) classmethod

Update the feature for the entire population (class method).

Initialize a random trial in the population if time is the random trial start time.

Parameters:

Name	Type	Description	Default
model	model.TITAN	the instance of TITAN currently being run	required

Source code in titan/features/random_trial.py

@classmethod
def update_pop(cls, model: "model.TITAN"):
    """
    Update the feature for the entire population (class method).

    Initialize a random trial in the population if time is the random trial start time.

    args:
        model: the instance of TITAN currently being run
    """
    rt_params = model.params.random_trial

    if not model.time == rt_params.start_time:
        return

    assert (
        model.params.model.network.enable
    ), "Network must be enabled for random trial"

    logging.info(f"Starting random trial ({rt_params.choice})")
    components = model.pop.connected_components()

    # set up helper methods based on params
    if rt_params.treatment == "prep":
        assert (
            model.params.features.prep
        ), "Prep feature must be enabled to use the prep random trial treatment"
        treat = treat_prep
        suitable = suitable_prep
    elif rt_params.treatment == "knowledge":
        assert (
            model.params.exposures.knowledge
        ), "Knowledge exposure must be enabled to use the knowledge random trial treatment"
        treat = treat_knowledge
        suitable = suitable_knowledge

    total_nodes = 0
    logging.info(
        f"Number of components {len([1 for comp in components if comp.number_of_nodes()])}",
    )
    for comp in components:
        total_nodes += comp.number_of_nodes()
        if model.run_random.random() < rt_params.prob:
            # Component selected as treatment pod!
            for agent in comp.nodes:
                agent.random_trial.active = True

            # treat all agents
            if rt_params.choice == "all":
                for agent in comp.nodes():
                    if suitable(agent, model):
                        treat(agent, model)
                        agent.random_trial.suitable = True
                        agent.random_trial.treated = True

            # chose an agent central to the component
            elif rt_params.choice == "eigenvector":
                centrality = nx.algorithms.centrality.eigenvector_centrality(comp)
                assert len(centrality) >= 1, "Empty centrality"
                ordered_centrality = sorted(centrality, key=centrality.get)

                # find the most central suitable agent, or if none, use most central
                intervention_agent = ordered_centrality[0]
                for agent in ordered_centrality:
                    if suitable(agent, model):
                        intervention_agent = agent
                        intervention_agent.random_trial.suitable = True
                        break

                intervention_agent.random_trial.treated = True
                treat(intervention_agent, model)

            # chose an agent that is a bridge in the network
            elif rt_params.choice == "bridge":
                # list all edges that are bridges
                all_bridges = list(nx.bridges(comp))
                suitable_agents = [
                    agent
                    for agents in all_bridges
                    for agent in agents
                    if suitable(agent, model)
                ]  # all suitable agents in bridges

                chosen_agent = utils.safe_random_choice(
                    suitable_agents, model.run_random
                )  # select change agent
                if chosen_agent is not None:
                    chosen_agent.random_trial.suitable = True  # type: ignore[attr-defined]

                else:  # if no suitable agents, mark a non-suitable agent
                    chosen_agent = utils.safe_random_choice(
                        list(comp.nodes), model.run_random
                    )

                chosen_agent.random_trial.treated = True  # type: ignore[attr-defined]
                treat(chosen_agent, model)

            # chose an agent from the component at random
            elif rt_params.choice == "random":
                suitable_agents = [
                    agent for agent in comp.nodes if suitable(agent, model)
                ]

                # if there are agents who meet eligibility criteria,
                # select one randomly
                chosen_agent = utils.safe_random_choice(
                    suitable_agents, model.run_random
                )

                if chosen_agent is not None:
                    chosen_agent.random_trial.suitable = True
                else:  # if no suitable agents, mark a non-suitable agent
                    chosen_agent = utils.safe_random_choice(
                        list(comp.nodes), model.run_random
                    )

                chosen_agent.random_trial.treated = True  # type: ignore[attr-defined]
                treat(chosen_agent, model)

    logging.info(f"Total agents in trial: {total_nodes}")