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 |
ag.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[ag.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 |
ag.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 |
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 |
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. |
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()