#
# Copyright 2024 John Harwell, All rights reserved.
#
# SPDX-License-Identifier: MIT
#
"""Inter-experiment linegraph generation in stage 4."""
# Core packages
import typing as tp
import logging
# 3rd party packages
import json
import yaml
import strictyaml
# Project packages
from sierra.core import types, batchroot, graphs
from sierra.core.graphs import bcbridge, schema
_logger = logging.getLogger(__name__)
[docs]
def generate(
cmdopts: types.Cmdopts,
pathset: batchroot.PathSet,
targets: list[types.YAMLDict],
info: bcbridge.GraphInfo,
) -> None:
"""Generate linegraphs from :term:`Collated Output Data` files.
The graphs generated by this module respect the ``--exp-range`` cmdline
option.
"""
_logger.info(
"Linegraphs from <batchroot>/%s",
pathset.stat_interexp_root.relative_to(pathset.root),
)
# For each category of linegraphs we are generating
for category in targets:
# For each graph in each category
for graph in category:
if graph["type"] not in ["summary_line", "stacked_line"]:
continue
_logger.trace("\n" + json.dumps(graph, indent=4))
if graph["type"] == "summary_line":
try:
loaded = strictyaml.load(yaml.dump(graph), schema.summary_line).data
except strictyaml.YAMLError as e:
_logger.critical("Non-conformant summary_line YAML: %s", e)
raise
_gen_summary_linegraph(loaded, pathset, cmdopts, info)
elif graph["type"] == "stacked_line":
try:
loaded = strictyaml.load(yaml.dump(graph), schema.stacked_line).data
except strictyaml.YAMLError as e:
_logger.critical("Non-conformant stacked_line YAML: %s", e)
raise
_gen_stacked_linegraph(loaded, pathset, cmdopts, info)
def _gen_summary_linegraph(
graph: types.YAMLDict,
pathset: batchroot.PathSet,
cmdopts: types.Cmdopts,
info: bcbridge.GraphInfo,
) -> None:
legend = "{}+{}".format(cmdopts["controller"], cmdopts["scenario"])
paths = graphs.PathSet(
input_root=pathset.stat_interexp_root,
output_root=pathset.graph_interexp_root,
batchroot=pathset.root,
model_root=pathset.model_interexp_root,
)
# 2025-06-05 [JRH]: We always write stage {3,4} output data files as .csv
# because that is currently SIERRA's 'native' format; this may change in the
# future.
graphs.summary_line(
paths=paths,
input_stem=graph["dest_stem"],
output_stem=graph["dest_stem"],
medium="storage.csv",
legend=[legend],
stats=cmdopts.get("dist_stats", "none"),
title=graph["title"],
xlabel=info.xlabel,
ylabel=graph.get("ylabel", None),
backend=graph.get("backend", cmdopts["graphs_backend"]),
xticks=info.xticks,
xticklabels=info.xticklabels,
logyscale=graph.get("logy", cmdopts["plot_log_yscale"]),
large_text=cmdopts["plot_large_text"],
)
def _gen_stacked_linegraph(
graph: types.YAMLDict,
pathset: batchroot.PathSet,
cmdopts: types.Cmdopts,
info: bcbridge.GraphInfo,
) -> None:
paths = graphs.PathSet(
input_root=pathset.stat_interexp_root,
output_root=pathset.graph_interexp_root,
model_root=None,
batchroot=pathset.root,
)
graphs.stacked_line(
paths=paths,
input_stem=graph["dest_stem"],
output_stem=graph["dest_stem"],
stats=cmdopts.get("dist_stats", "none"),
medium="storage.csv",
title=graph["title"],
backend=graph.get("backend", cmdopts["graphs_backend"]),
xticks=None,
xlabel=graph.get("xlabel", "Time"),
ylabel=graph.get("ylabel", None),
logyscale=graph.get("logy", cmdopts["plot_log_yscale"]),
large_text=cmdopts["plot_large_text"],
legend=graph.get("legend", [f"exp{i}" for i in range(0, len(info.exp_names))]),
)
__all__ = ["generate"]
