Skip to content

Reference

plot_data

A set of functions to get the data series to plot for test run results.

split_metric_uncertainty(metrics: dict[str, str | UFloat], metric: str) -> tuple[float, float | None]

Get the uncertainty and value from a possible uncertain metric.

Source code in src/hpc_multibench/plot/plot_data.py 
13
14
15
16
17
18
19
20
def split_metric_uncertainty(
    metrics: dict[str, str | UFloat], metric: str
) -> tuple[float, float | None]:
    """Get the uncertainty and value from a possible uncertain metric."""
    value = metrics[metric]
    if isinstance(value, UFloat):
        return (value.nominal_value, value.std_dev)
    return (float(value), None)

get_line_plot_data(plot: LinePlotModel, all_metrics: list[tuple[RunConfiguration, dict[str, str | UFloat]]]) -> dict[str, tuple[list[float], list[float], list[float] | None, list[float] | None]]

Get the data needed to plot a specified line plot for a set of runs.

Source code in src/hpc_multibench/plot/plot_data.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def get_line_plot_data(
    plot: LinePlotModel,
    all_metrics: list[tuple[RunConfiguration, dict[str, str | UFloat]]],
) -> dict[str, tuple[list[float], list[float], list[float] | None, list[float] | None]]:
    """Get the data needed to plot a specified line plot for a set of runs."""
    # Reshape the metrics data from multiple runs into groups of points
    data: dict[str, list[tuple[float, float, float | None, float | None]]] = {}
    for run_configuration, metrics in all_metrics:
        split_names: list[str] = [
            f"{split_metric}={metrics[split_metric]}"
            for split_metric in plot.split_metrics
            if split_metric not in plot.fix_metrics
        ]
        fix_names: list[str] = [
            f"{metric}={value}" for metric, value in plot.fix_metrics.items()
        ]
        series_identifier = ", ".join(
            [run_configuration.name, *fix_names, *split_names]
        )

        if any(
            metrics[metric] != str(value) for metric, value in plot.fix_metrics.items()
        ):
            continue

        (x_value, x_err) = split_metric_uncertainty(metrics, plot.x)

        y_metrics = plot.y if isinstance(plot.y, list) else [plot.y]
        for y_metric in y_metrics:
            series_name = (
                f"{y_metric}, {series_identifier}"
                if isinstance(plot.y, list)
                else series_identifier
            )
            (y_value, y_err) = split_metric_uncertainty(metrics, y_metric)
            if series_name not in data:
                data[series_name] = []
            data[series_name].append((x_value, y_value, x_err, y_err))

    # Further reshape the data into convenient data series
    reshaped_data: dict[
        str, tuple[list[float], list[float], list[float] | None, list[float] | None]
    ] = {}
    for name, results in data.items():
        x, y, x_err, y_err = cast(  # type: ignore[assignment]
            tuple[list[float], list[float], list[float | None], list[float | None]],
            zip(*sorted(results), strict=True),
        )
        reshaped_data[name] = (  # type: ignore[assignment]
            x,
            y,
            x_err if any(x_err) else None,  # type: ignore[arg-type]
            y_err if any(y_err) else None,  # type: ignore[arg-type]
        )
    return reshaped_data

get_bar_chart_data(plot: BarChartModel, all_metrics: list[tuple[RunConfiguration, dict[str, str | UFloat]]]) -> dict[str, tuple[float, float | None, int]]

Get the data needed to plot a specified bar chart for a set of runs.

Source code in src/hpc_multibench/plot/plot_data.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def get_bar_chart_data(
    plot: BarChartModel,
    all_metrics: list[tuple[RunConfiguration, dict[str, str | UFloat]]],
) -> dict[str, tuple[float, float | None, int]]:
    """Get the data needed to plot a specified bar chart for a set of runs."""
    data: dict[str, tuple[float, float | None, int]] = {}

    # Extract the outputs into the data format needed for the line plot
    hue_index_lookup: dict[str, int] = {}
    new_hue_index = 0
    for run_configuration, metrics in all_metrics:
        split_names: list[str] = [
            f"{split_metric}={metrics[split_metric]}"
            for split_metric in plot.split_metrics
            if split_metric not in plot.fix_metrics
        ]
        fix_names: list[str] = [
            f"{metric}={value}" for metric, value in plot.fix_metrics.items()
        ]
        series_identifier = ", ".join(
            [run_configuration.name, *fix_names, *split_names]
        )

        if any(
            metrics[metric] != str(value) for metric, value in plot.fix_metrics.items()
        ):
            continue

        if run_configuration.name not in hue_index_lookup:
            hue_index_lookup[run_configuration.name] = new_hue_index
            new_hue_index += 1

        y_metrics = plot.y if isinstance(plot.y, list) else [plot.y]
        for y_metric in y_metrics:
            series_name = (
                f"{y_metric}, {series_identifier}"
                if isinstance(plot.y, list)
                else series_identifier
            )
            (y_value, y_err) = split_metric_uncertainty(metrics, y_metric)
            data[series_name] = (
                y_value,
                y_err,
                hue_index_lookup[run_configuration.name],
            )
    return data

get_roofline_plot_data(plot: RooflinePlotModel, all_metrics: list[tuple[RunConfiguration, dict[str, str | UFloat]]]) -> tuple[RooflineDataModel, dict[str, tuple[float, float, float | None, float | None]]]

Get the data needed to plot a specified roofline plot.

Source code in src/hpc_multibench/plot/plot_data.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def get_roofline_plot_data(
    plot: RooflinePlotModel,
    all_metrics: list[tuple[RunConfiguration, dict[str, str | UFloat]]],
) -> tuple[
    RooflineDataModel, dict[str, tuple[float, float, float | None, float | None]]
]:
    """Get the data needed to plot a specified roofline plot."""
    roofline_data = RooflineDataModel.from_json(plot.ert_json)

    data: dict[str, tuple[float, float, float | None, float | None]] = {}
    for run_configuration, metrics in all_metrics:
        (y_value, y_err) = split_metric_uncertainty(metrics, plot.gflops_per_sec)
        (x_value_tmp, x_err_tmp) = split_metric_uncertainty(
            metrics, plot.mbytes_per_sec
        )
        x_value = y_value / (x_value_tmp / 1000)
        x_err = None if x_err_tmp is None or y_err is None else y_err + x_err_tmp
        data[run_configuration.name] = (x_value, y_value, x_err, y_err)

    return (roofline_data, data)

plot_matplotlib

A set of functions using matplotlib to plot the results of a test bench run.

draw_line_plot(plot: LinePlotModel, metrics: list[tuple[RunConfiguration, dict[str, str | UFloat]]]) -> None

Draw a specified line plot for a set of run outputs.

Source code in src/hpc_multibench/plot/plot_matplotlib.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def draw_line_plot(
    plot: LinePlotModel,
    metrics: list[tuple[RunConfiguration, dict[str, str | UFloat]]],
) -> None:
    """Draw a specified line plot for a set of run outputs."""
    data = get_line_plot_data(plot, metrics)

    for name, (x, y, x_err, y_err) in data.items():
        plt.errorbar(
            x,
            y,
            xerr=x_err,
            yerr=y_err,
            marker="x",
            ecolor="black",
            label=name,
        )
    plt.legend()
    plt.xlabel(plot.x)
    if isinstance(plot.y, list):
        plt.ylabel(" | ".join(plot.y))
    else:
        plt.ylabel(plot.y)
    plt.title(plot.title)
    if plot.x_lim is not None:
        plt.xlim(plot.x_lim)
    if plot.y_lim is not None:
        plt.ylim(plot.y_lim)
    if plot.x_log:
        plt.xscale("log")
    if plot.y_log:
        plt.yscale("log")
    plt.show()

draw_bar_chart(plot: BarChartModel, metrics: list[tuple[RunConfiguration, dict[str, str | UFloat]]]) -> None

Draw a specified bar chart for a set of run outputs.

Source code in src/hpc_multibench/plot/plot_matplotlib.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def draw_bar_chart(
    plot: BarChartModel,
    metrics: list[tuple[RunConfiguration, dict[str, str | UFloat]]],
) -> None:
    """Draw a specified bar chart for a set of run outputs."""
    data = get_bar_chart_data(plot, metrics)

    # For matplotlib `plt.rcParams["axes.prop_cycle"].by_key()["color"]`
    palette = sns.color_palette()
    plt.barh(
        list(data.keys()),
        [metric for metric, _, _ in data.values()],
        xerr=[uncertainty for _, uncertainty, _ in data.values()],
        color=[palette[hue] for _, _, hue in data.values()],
        ecolor="black",
    )
    plt.xlabel(plot.y)
    plt.gcf().subplots_adjust(left=0.25)
    plt.title(plot.title)
    if plot.y_lim is not None:
        plt.ylim(plot.y_lim)
    if plot.y_log:
        plt.yscale("log")
    plt.show()

draw_roofline_plot(plot: RooflinePlotModel, metrics: list[tuple[RunConfiguration, dict[str, str | UFloat]]]) -> None

Draw a specified roofline plots for a set of run outputs.

Source code in src/hpc_multibench/plot/plot_matplotlib.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def draw_roofline_plot(
    plot: RooflinePlotModel,
    metrics: list[tuple[RunConfiguration, dict[str, str | UFloat]]],
) -> None:
    """Draw a specified roofline plots for a set of run outputs."""
    (roofline, data) = get_roofline_plot_data(plot, metrics)

    for label, (x, y) in roofline.memory_bound_ceilings.items():
        plt.plot(x, y, label=label)
    for label, (x, y) in roofline.compute_bound_ceilings.items():
        plt.plot(x, y, label=label)
    # from labellines import labelLines
    # for ax in plt.gcf().axes:
    #     labelLines(ax.get_lines())
    for name, (x_point, y_point, x_err, y_err) in data.items():
        plt.errorbar(
            x_point,
            y_point,
            xerr=x_err,
            yerr=y_err,
            marker="o",
            ecolor="black",
            label=name,
        )
    plt.legend()
    plt.xlabel("FLOPs/Byte")
    plt.ylabel("GFLOPs/sec")
    plt.xscale("log")
    plt.yscale("log")
    plt.title(plot.title)
    plt.show()

plot_plotext

A set of functions using plotext to plot the results of a test bench run.

draw_line_plot(this_plt: Any, plot: LinePlotModel, metrics: list[tuple[RunConfiguration, dict[str, str | UFloat]]]) -> None

Draw a line plot from data using the provided plotext backend.

Source code in src/hpc_multibench/plot/plot_plotext.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def draw_line_plot(
    this_plt: Any,
    plot: LinePlotModel,
    metrics: list[tuple[RunConfiguration, dict[str, str | UFloat]]],
) -> None:
    """Draw a line plot from data using the provided plotext backend."""
    data = get_line_plot_data(plot, metrics)

    this_plt.clear_figure()
    for name, (x, y, _x_err, _y_err) in data.items():
        this_plt.plot(x, y, marker=PLOTEXT_MARKER, label=name)
    this_plt.theme(PLOTEXT_THEME)
    this_plt.xlabel(plot.x)
    if isinstance(plot.y, list):
        this_plt.ylabel(" | ".join(plot.y))
    else:
        this_plt.ylabel(plot.y)
    this_plt.ylim(0)
    if plot.x_log:
        this_plt.xscale("log")
    if plot.y_log:
        this_plt.yscale("log")
    this_plt.title(plot.title)

draw_bar_chart(this_plt: Any, plot: BarChartModel, metrics: list[tuple[RunConfiguration, dict[str, str | UFloat]]]) -> None

Draw a line plot from data using the provided plotext backend.

Source code in src/hpc_multibench/plot/plot_plotext.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def draw_bar_chart(
    this_plt: Any,
    plot: BarChartModel,
    metrics: list[tuple[RunConfiguration, dict[str, str | UFloat]]],
) -> None:
    """Draw a line plot from data using the provided plotext backend."""
    data = get_bar_chart_data(plot, metrics)

    this_plt.clear_figure()
    this_plt.bar(
        data.keys(),
        [metric for metric, _, _ in data.values()],
        orientation="horizontal",
        width=3 / 5,
    )
    this_plt.theme(PLOTEXT_THEME)
    this_plt.ylabel(plot.y)
    this_plt.title(plot.title)
    if plot.y_log:
        this_plt.yscale("log")

draw_roofline_plot(this_plt: Any, plot: RooflinePlotModel, metrics: list[tuple[RunConfiguration, dict[str, str | UFloat]]]) -> None

Draw a roofline plot from data using the provided plotext backend.

Source code in src/hpc_multibench/plot/plot_plotext.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def draw_roofline_plot(
    this_plt: Any,
    plot: RooflinePlotModel,
    metrics: list[tuple[RunConfiguration, dict[str, str | UFloat]]],
) -> None:
    """Draw a roofline plot from data using the provided plotext backend."""
    (roofline, data) = get_roofline_plot_data(plot, metrics)

    this_plt.clear_figure()
    for label, (x, y) in roofline.memory_bound_ceilings.items():
        this_plt.plot(x, y, label=label, marker=PLOTEXT_MARKER)
    for label, (x, y) in roofline.compute_bound_ceilings.items():
        this_plt.plot(x, y, label=label, marker=PLOTEXT_MARKER)
    for name, (x_point, y_point, x_err, y_err) in data.items():
        this_plt.error(
            [x_point],
            [y_point],
            xerr=[x_err / 2] * 2 if x_err is not None else None,
            yerr=[y_err / 2] * 2 if y_err is not None else None,
            label=name,
        )
    this_plt.theme(PLOTEXT_THEME)
    this_plt.xlabel("FLOPs/Byte")
    this_plt.ylabel("GFLOPs/sec")
    this_plt.xscale("log")
    this_plt.yscale("log")
    this_plt.title(plot.title)

export_data

A set of functions to export the results of a test bench run.

export_data(plot: ExportModel, all_metrics: list[tuple[RunConfiguration, dict[str, str | UFloat]]]) -> None

Construct and export a pandas data frame from the metrics.

Source code in src/hpc_multibench/plot/export_data.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def export_data(
    plot: ExportModel,
    all_metrics: list[tuple[RunConfiguration, dict[str, str | UFloat]]],
) -> None:
    """Construct and export a pandas data frame from the metrics."""
    df_data: dict[str, list[float | str]] = {}
    for run_configuration, metrics in all_metrics:
        row_data: dict[str, float | str] = {"Run configuration": run_configuration.name}

        for metric in metrics:
            value, error = split_metric_uncertainty(metrics, metric)
            row_data[metric] = value
            if error is not None:
                row_data[f"{metric} error"] = error

        for column, cell in row_data.items():
            if column not in df_data:
                df_data[column] = []
            df_data[column].append(cell)

    export_df = pd.DataFrame(df_data)

    if plot.export_path is None:
        print(export_df.to_string())
    elif plot.export_format == "csv":
        export_df.to_csv(plot.export_path)
    elif plot.export_format == "latex":
        export_df.to_latex(plot.export_path)
    else:
        raise NotImplementedError(
            f"Export format '{plot.export_format}' not supported!"
        )