Reporting

add_agent_to_stats(stats_item, key)

Update the stats dictionary counts for the key given the agent's attributes

Parameters:

Name	Type	Description	Default
stats_item	Dict[str, int]	the leaf node of a nested dictionary of attributes to counts	required
key	str	the type of count to increment	required

Source code in titan/output.py

def add_agent_to_stats(stats_item: Dict[str, int], key: str):
    """
    Update the stats dictionary counts for the key given the agent's attributes

    args:
        stats_item: the leaf node of a nested dictionary of attributes to counts
        key: the type of count to increment
    """
    stats_item[key] += 1

basicReport(run_id, t, runseed, popseed, stats, params, outdir)

Standard report writer for basic agent statistics, columns include:

• "agents": number of agents in the population
• "[exit]": number of agents who exited this time period by exit class
• "[exit]_hiv": number of agents with HIV who exited this time period by exit class

Additionally, any feature enabled may have additional stats that are tracked. See the feature's stats attribute and docs for details.

Source code in titan/output.py

def basicReport(
    run_id: str,
    t: int,
    runseed: int,
    popseed: int,
    stats: Dict[str, Any],
    params: ObjMap,
    outdir: str,
):
    """
    Standard report writer for basic agent statistics, columns include:

    * "agents": number of agents in the population
    * "[exit]": number of agents who exited this time period by exit class
    * "[exit]_hiv": number of agents with HIV who exited this time period by exit class

    Additionally, any feature enabled may have additional stats that are tracked.  See the feature's `stats` attribute and docs for details.
    """
    write_report("basicReport.txt", run_id, t, runseed, popseed, stats, params, outdir)

get_agg_val(stats, attrs, key)

Get the value of a key in stats given the attribute values

Parameters:

Name	Type	Description	Default
stats	Dict	a nested dictionary of attributes to count	required
attrs	List	a list of attribute values to find the count for	required
key	str	the type of count to get the value of	required

Returns:

Type	Description
int	the count of key for the given attributes

Source code in titan/output.py

def get_agg_val(stats: Dict, attrs: List, key: str) -> int:
    """
    Get the value of a key in stats given the attribute values

    args:
        stats: a nested dictionary of attributes to count
        attrs: a list of attribute values to find the count for
        key: the type of count to get the value of

    returns:
        the count of key for the given attributes
    """
    stats_item = stats
    for attr in attrs:
        stats_item = stats_item[attr]

    return stats_item[key]

get_aggregates(params)

Get iterator over all attribute combinations for output classes

Parameters:

Name	Type	Description	Default
params	ObjMap	model parameters	required

Returns:

Type	Description
Iterator	iterator over attribute combinations

Source code in titan/output.py

def get_aggregates(params: ObjMap) -> Iterator:
    """
    Get iterator over all attribute combinations for output classes

    args:
        params: model parameters

    returns:
        iterator over attribute combinations
    """
    return itertools.product(
        *[list(k for k in params.classes[clss]) for clss in params.outputs.classes]
    )

get_stats(all_agents, exits, params, exposures, features, time)

Get the current statistics for a model based on the population, and tracking agent sets from the model.

Parameters:

Name	Type	Description	Default
all_agents	AgentSet	all of the agents in the population	required
new_hiv.dx		agents who are newly diagnosed with hiv this timestep	required
exits	Dict[str, List[Agent]]	dictionary indicated how many agents exited by exit class this timestep	required
params	ObjMap	model parameters	required

Returns:

Type	Description
Dict	nested dictionary of agent attributes to counts of various items

Source code in titan/output.py

def get_stats(
    all_agents: "ag.AgentSet",
    exits: Dict[str, List["ag.Agent"]],
    params: ObjMap,
    exposures,
    features,
    time: int,
) -> Dict:
    """
    Get the current statistics for a model based on the population, and tracking agent sets from the model.

    args:
        all_agents: all of the agents in the population
        new_hiv.dx: agents who are newly diagnosed with hiv this timestep
        exits: dictionary indicated how many agents exited by exit class this timestep
        params: model parameters

    returns:
        nested dictionary of agent attributes to counts of various items
    """
    reportables = exposures + features
    stats = setup_aggregates(params, reportables, params.outputs.classes, exits)

    # attribute names (non-plural)
    attrs = [clss[:-1] for clss in params.outputs.classes]

    for a in all_agents:
        stats_item = get_stats_item(stats, attrs, a)

        add_agent_to_stats(stats_item, "agents")

        for reportable in reportables:
            agent_feature = getattr(a, reportable.name)
            agent_feature.set_stats(stats_item, time)

    for exit in exits:
        for a in exits[exit]:
            stats_item = get_stats_item(stats, attrs, a)
            add_agent_to_stats(stats_item, exit)
            if a.hiv.active:  # type: ignore[attr-defined]
                add_agent_to_stats(stats_item, exit + "_hiv")

    return stats

get_stats_item(stats, attrs, agent)

Get the leaf node of the stats dictionary for the given attributes and agent.

Parameters:

Name	Type	Description	Default
stats	Dict[str, Any]	a nested dictionary of attributes to count	required
attrs	List[str]	a list of attribute values to find the count for	required
agent	Agent	The agent to get the leaf node for	required

Returns:

Type	Description
	a stats_item dictionary of keys to counts

Source code in titan/output.py

def get_stats_item(stats: Dict[str, Any], attrs: List[str], agent: "ag.Agent"):
    """
    Get the leaf node of the stats dictionary for the given attributes and agent.

    args:
        stats: a nested dictionary of attributes to count
        attrs: a list of attribute values to find the count for
        agent: The agent to get the leaf node for

    returns:
        a stats_item dictionary of keys to counts
    """
    stats_item = stats
    for attr in attrs:
        stats_item = stats_item[str(getattr(agent, attr))]

    return stats_item

print_components(run_id, t, runseed, popseed, components, outdir)

Write stats describing the components (sub-graphs) in a graph to file

Parameters:

Name	Type	Description	Default
run_id	str	unique identifer for this run of the model	required
t	int	current timestep	required
runseed	int	integer used to seed the model's random number generator	required
popseed	int	integer used to seed the population's random number generator	required
components	List	a list of graph components	required
outdir	str	path where the file should be saved	required
races		the races in the population	required

Source code in titan/output.py

def print_components(
    run_id: str,
    t: int,
    runseed: int,
    popseed: int,
    components: List,
    outdir: str,
):
    """
    Write stats describing the components (sub-graphs) in a graph to file

    args:
        run_id: unique identifer for this run of the model
        t: current timestep
        runseed: integer used to seed the model's random number generator
        popseed: integer used to seed the population's random number generator
        components: a list of graph components
        outdir: path where the file should be saved
        races: the races in the population
    """
    f = open(os.path.join(outdir, f"{run_id}_componentReport_ALL.txt"), "a")

    # if this is a new file, write the header info
    if f.tell() == 0:
        f.write(
            "run_id\trunseed\tpopseed\tt\tcomponent\tagents"
            "\tinfected\ttrt\ttrtinfected"
            "\tdensity\tEffectiveSize\tdeg_cent\n"
        )

    for id, comp in enumerate(components):
        num_nodes = comp.number_of_nodes()

        average_size = effective_size(comp) / num_nodes
        comp_density = nx.density(comp)

        deg_cent = mean(list(nx.degree_centrality(comp).values()))
        nhiv = ntrthiv = ntrt = 0
        for agent in comp.nodes():
            if agent.hiv.active:
                nhiv += 1
                if agent.random_trial.treated:
                    ntrthiv += 1

            if agent.random_trial.treated:
                ntrt += 1

        f.write(
            f"{run_id}\t{runseed}\t{popseed}\t{t}\t{id}\t{num_nodes}"
            f"\t{nhiv}\t{ntrt}\t{ntrthiv}"
            f"\t{comp_density:.4f}"
            f"\t{average_size:.4f}\t{deg_cent}\n"
        )

    f.close()

setup_aggregates(params, reportables, classes, exits)

Recursively create a nested dictionary of attribute values to items to count.

Attributes are classes defined in params, the items counted are:

• "agents"
• "[exit_class]_hiv"
• exit by exit class

Additionally, any feature enabled may have additional stats that are tracked. See the feature's stats attribute.

Parameters:

Name	Type	Description	Default
params	ObjMap	model parameters	required
classes	List[str]	which classes to aggregate by [params.outputs.classes]	required

Returns:

Type	Description
Dict	dictionary of class values to counts

Source code in titan/output.py

def setup_aggregates(
    params: ObjMap, reportables, classes: List[str], exits: Dict[str, List["ag.Agent"]]
) -> Dict:
    """
    Recursively create a nested dictionary of attribute values to items to count.

    Attributes are classes defined in params, the items counted are:

    * "agents"
    * "[exit_class]_hiv"
    * exit by exit class

    Additionally, any feature enabled may have additional stats that are tracked.  See the feature's `stats` attribute.

    args:
        params: model parameters
        classes: which classes to aggregate by [params.outputs.classes]

    returns:
        dictionary of class values to counts
    """
    if classes == []:
        base_stats = {
            "agents": 0,
        }
        for exit in params.classes.exit.keys():
            base_stats[exit] = 0
            base_stats[exit + "_hiv"] = 0

        for reportable in reportables:
            base_stats.update({stat: 0 for stat in reportable.stats})

        return base_stats

    stats = {}
    clss, *rem_clss = classes  # head, tail
    keys = [k for k in params.classes[clss]]

    for key in keys:
        stats[key] = setup_aggregates(params, reportables, rem_clss, exits)

    return stats

write_graph_edgelist(graph, path, id, time)

Writes a pipe-delimited edge list to the file <id>_Edgelist_t<time>.txt

Parameters:

Name	Type	Description	Default
path	str	directory where the file should be saved	required
id		identifier for the network, typically the model's id	required
time		timestep the edgelist is being written at	required

Source code in titan/output.py

def write_graph_edgelist(graph, path: str, id, time):
    """
    Writes a pipe-delimited edge list to the file `<id>_Edgelist_t<time>.txt`

    args:
        path: directory where the file should be saved
        id: identifier for the network, typically the model's `id`
        time: timestep the edgelist is being written at
    """
    file_path = os.path.join(path, f"{id}_Edgelist_t{time}.txt")
    # Write edgelist with bond type
    nx.write_edgelist(graph, file_path, delimiter="|", data=["type"])

write_network_stats(graph, path, id, time)

Writes network statistics to the file <id>_NetworkStats_t<time>.txt

Parameters:

Name	Type	Description	Default
path	str	directory where the file should be saved	required
id		identifier for the network, typically the model's id	required
time		timestep the edgelist is being written at	required

Source code in titan/output.py

def write_network_stats(graph, path: str, id, time):
    """
    Writes network statistics to the file `<id>_NetworkStats_t<time>.txt`

    args:
        path: directory where the file should be saved
        id: identifier for the network, typically the model's `id`
        time: timestep the edgelist is being written at
    """
    file_path = os.path.join(path, f"{id}_NetworkStats_t{time}.txt")

    components = utils.connected_components(graph)

    outfile = open(file_path, "w")
    outfile.write(nx.info(graph))

    cent_dict = nx.degree_centrality(graph)

    outfile.write(
        "\nNumber of connected components: {}\n".format(
            nx.number_connected_components(graph)
        )
    )

    tot_nodes = 0
    for c in components:
        tot_nodes += c.number_of_nodes()

    outfile.write(f"Number of nodes: {tot_nodes}\n")

    outfile.write(
        "Average component size: {}\n".format(
            tot_nodes * 1.0 / nx.number_connected_components(graph)
        )
    )
    outfile.write(
        "Maximum component size: {}\n".format(nx.number_of_nodes(components[0]))
    )
    outfile.write("Degree Histogram: {}\n".format(nx.degree_histogram(graph)))
    outfile.write("Graph density: {}\n".format(nx.density(graph)))
    outfile.write(
        "Average node degree centrality: {}\n".format(
            sum(cent_dict.values()) / len(list(cent_dict.values()))
        )
    )

    outfile.write("Average node clustering: {}\n".format(nx.average_clustering(graph)))
    outfile.close()

write_report(file_name, run_id, t, runseed, popseed, stats, params, outdir)

Core function for writing reports, writes header if file is new, then data based on the params and name_map

Parameters:

Name	Type	Description	Default
file_name	str	Name of the file to write, including the extension (e.g. MyReport.txt)	required
run_id	str	unique identifier for this model	required
t	int	current timestep	required
runseed	int	integer used to seed the random number generator for the model	required
popseed	int	integer used to seed the random number generator for the population	required
stats	Dict	nested dictionary of agent attributes to counts	required
params	ObjMap	model parameters	required
outdir	str	path of where to save this file	required

Source code in titan/output.py

def write_report(
    file_name: str,
    run_id: str,
    t: int,
    runseed: int,
    popseed: int,
    stats: Dict,
    params: ObjMap,
    outdir: str,
):
    """
    Core function for writing reports, writes header if file is new, then data based
    on the `params` and `name_map`

    args:
        file_name: Name of the file to write, including the extension (e.g. `MyReport.txt`)
        run_id: unique identifier for this model
        t: current timestep
        runseed: integer used to seed the random number generator for the model
        popseed: integer used to seed the random number generator for the population
        stats: nested dictionary of agent attributes to counts
        params: model parameters
        outdir: path of where to save this file
    """

    def get_stat_names(stats, attrs):
        stat_ref = stats
        for i in range(len(attrs)):
            stat_ref = stat_ref[list(stat_ref.keys())[0]]

        return stat_ref

    f = open(os.path.join(outdir, file_name), "a")
    attrs = [clss[:-1] for clss in params.outputs.classes]
    stat_names = get_stat_names(stats, attrs)

    if f.tell() == 0:
        f.write("run_id\trseed\tpseed\tt\t")  # start header

        # attributes in stats
        f.write("\t".join(attrs))

        # report specific fields
        for name in stat_names:
            f.write(f"\t{name}")

        f.write("\n")

    for agg in get_aggregates(params):
        # don't write row if no agents are in it
        if get_agg_val(stats, agg, "agents") > 0:
            f.write(f"{run_id}\t{runseed}\t{popseed}\t{t}\t")

            f.write("\t".join(agg))  # write attribute values

            for name in stat_names:
                f.write(f"\t{(get_agg_val(stats, agg, name))}")

            f.write("\n")

    f.close()