Source code for sdom.resiliency.plotting
"""Distribution plots for :class:`sdom.resiliency.ResiliencyResults`.
Phase 6 deliverable C. Matplotlib is imported lazily so the module can be
imported in head-less environments without a display.
"""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Any
import numpy as np
if TYPE_CHECKING: # pragma: no cover - typing only
from matplotlib.axes import Axes
from sdom.resiliency.system_state import ResiliencyResults
logger = logging.getLogger(__name__)
_VALID_KINDS = ("hist", "ecdf", "exceedance")
__all__ = ["plot_metric_distribution"]
[docs]
def plot_metric_distribution(
results: "ResiliencyResults",
*,
metric: str = "EUE",
kind: str = "hist",
ax: "Axes | None" = None,
**plot_kwargs: Any,
) -> "Axes":
"""Plot the empirical distribution of a per-hour metric.
Parameters
----------
results : ResiliencyResults
Container produced by :func:`sdom.resiliency.run_resiliency_evaluation`.
metric : str, optional
Numeric column of ``results.per_hour`` to plot. Default ``"EUE"``.
kind : {"hist", "ecdf", "exceedance"}, optional
Plot style. Default ``"hist"``.
* ``"hist"`` - histogram of metric values.
* ``"ecdf"`` - empirical CDF, monotonically non-decreasing in ``[0, 1]``.
* ``"exceedance"`` - exceedance curve ``1 - ECDF``, monotonically
non-increasing.
ax : matplotlib.axes.Axes, optional
Existing axes to draw on. A new figure/axes is created when ``None``.
**plot_kwargs
Forwarded to the underlying matplotlib call (``ax.hist`` for
``kind="hist"``; ``ax.plot`` otherwise).
Returns
-------
matplotlib.axes.Axes
Raises
------
ImportError
If matplotlib is not importable.
ValueError
If ``kind`` is not a supported value or ``metric`` is not a numeric
column of ``results.per_hour``.
"""
if kind not in _VALID_KINDS:
raise ValueError(
f"Invalid kind={kind!r}. Expected one of {_VALID_KINDS}."
)
try:
import matplotlib.pyplot as plt
except ImportError as exc: # pragma: no cover - matplotlib is a hard dep
raise ImportError(
"plot_metric_distribution requires matplotlib. Install with "
"'pip install matplotlib'."
) from exc
df = results.per_hour
if "solver_status" in df.columns:
n_total = len(df)
df = df[df["solver_status"] != "error"]
n_dropped = n_total - len(df)
if n_dropped:
logger.debug(
"plot_metric_distribution: dropping %d errored row(s) before plotting.",
n_dropped,
)
if metric not in df.columns:
numeric_cols = sorted(
c for c in df.columns if np.issubdtype(df[c].dtype, np.number)
)
raise ValueError(
f"Unknown metric={metric!r}. Available numeric columns: {numeric_cols}."
)
values = df[metric].astype(float).to_numpy()
values = values[~np.isnan(values)]
logger.info(
"plot_metric_distribution: metric=%r, kind=%r, n_points=%d.",
metric,
kind,
len(values),
)
if ax is None:
_, ax = plt.subplots()
if kind == "hist":
ax.hist(values, **plot_kwargs)
if not ax.get_ylabel():
ax.set_ylabel("count")
else:
x = np.sort(values)
n = len(x)
if n == 0:
y = np.array([], dtype=float)
else:
y = np.arange(1, n + 1, dtype=float) / n
if kind == "exceedance":
y = 1.0 - y + (1.0 / n if n else 0.0)
ax.plot(x, y, **plot_kwargs)
if not ax.get_ylabel():
ax.set_ylabel(
"P(X \u2264 x)" if kind == "ecdf" else "P(X > x)"
)
if not ax.get_xlabel():
ax.set_xlabel(metric)
return ax