k4bench.analysis.plots

k4bench.analysis.plots

Plotly plotting functions for k4bench results.

All functions return a :class:~plotly.graph_objects.Figure. In Jupyter the figure renders inline automatically; call fig.show() to display it explicitly, or fig.write_html("out.html") to export.

Typical notebook usage::

from k4bench.analysis import load_results, plot_run_overview, plot_event_timing

plot_run_overview("logs/")
plot_event_timing("logs/")

BinCountOptions

Bases: NamedTuple

Automatic and allowed dashboard bin counts for one data selection.

Measure dataclass

Measure(column: str, title: str, unit: str, write: Callable[[float], str], total: str | None = None, tickformat: str = ',~g', log: bool = True)

One measure a :func:plot_standing panel draws.

auto_bin_count

auto_bin_count(source: dict[str, DataFrame] | str | Path | list[str | Path], *, column: str, labels: list[str] | None = None, exclude_events: list[int] | None = None, outlier_threshold: float = 3.5) -> int

Return the bin count plot_event_* would choose with bins="auto".

Runs the same preparation the plotting functions do — the same event exclusions, the same outlier clipping, the same pooled range — so callers that want to offer an editable bin count can seed it with the number the figure would otherwise have used, instead of an arbitrary default that would silently rebin the plot on first render.

Parameters:

Name	Type	Description	Default
source	dict[str, pd.DataFrame], str/Path, or list of str/Path	As for :func:plot_event_timing.	required
column	str	The value column to bin, e.g. "event_time_s" or "rss_end_mb".	required
labels	list[str] or None	Restrict to these run labels.	None
exclude_events	list[int] or None	Event numbers to exclude. Defaults to [0].	None
outlier_threshold	float	MAD-based modified Z-score threshold for range clipping.	3.5

Source code in k4bench/analysis/plots/event.py

def auto_bin_count(
    source: dict[str, pd.DataFrame] | str | Path | list[str | Path],
    *,
    column: str,
    labels: list[str] | None = None,
    exclude_events: list[int] | None = None,
    outlier_threshold: float = 3.5,
) -> int:
    """Return the bin count ``plot_event_*`` would choose with ``bins="auto"``.

    Runs the same preparation the plotting functions do — the same event
    exclusions, the same outlier clipping, the same pooled range — so callers
    that want to offer an editable bin count can seed it with the number the
    figure would otherwise have used, instead of an arbitrary default that
    would silently rebin the plot on first render.

    Parameters
    ----------
    source : dict[str, pd.DataFrame], str/Path, or list of str/Path
        As for :func:`plot_event_timing`.
    column : str
        The value column to bin, e.g. ``"event_time_s"`` or ``"rss_end_mb"``.
    labels : list[str] or None
        Restrict to these run labels.
    exclude_events : list[int] or None
        Event numbers to exclude.  Defaults to ``[0]``.
    outlier_threshold : float
        MAD-based modified Z-score threshold for range clipping.
    """
    return event_bin_options(
        source,
        column=column,
        labels=labels,
        exclude_events=exclude_events,
        outlier_threshold=outlier_threshold,
    ).automatic

event_bin_options

event_bin_options(source: dict[str, DataFrame] | str | Path | list[str | Path], *, column: str, labels: list[str] | None = None, exclude_events: list[int] | None = None, outlier_threshold: float = 3.5) -> BinCountOptions

Return automatic and sensible editable bin counts for an event metric.

The maximum is derived from the current pooled in-range sample: more bins than observations normally add no statistical resolution. If NumPy's automatic rule itself chooses more, that exact automatic count remains selectable. The renderer ceiling remains the absolute upper bound.

Source code in k4bench/analysis/plots/event.py

def event_bin_options(
    source: dict[str, pd.DataFrame] | str | Path | list[str | Path],
    *,
    column: str,
    labels: list[str] | None = None,
    exclude_events: list[int] | None = None,
    outlier_threshold: float = 3.5,
) -> BinCountOptions:
    """Return automatic and sensible editable bin counts for an event metric.

    The maximum is derived from the current pooled in-range sample: more bins
    than observations normally add no statistical resolution. If NumPy's
    automatic rule itself chooses more, that exact automatic count remains
    selectable. The renderer ceiling remains the absolute upper bound.
    """
    if exclude_events is None:
        exclude_events = list(_DEFAULT_EXCLUDE_EVENTS)
    all_event_data = _ensure_event_data(source)
    if labels is None:
        event_data = all_event_data
    else:
        event_data = {
            key: frame
            for key, frame in all_event_data.items()
            if key in labels or any(key.endswith(f"/{wanted}") for wanted in labels)
        }
    if not event_data:
        raise ValueError(f"No event data found for labels={labels}.")
    prepared = _prepare_event_arrays(event_data, column, exclude_events, outlier_threshold)
    automatic = len(resolve_bin_edges(prepared.clipped_all)) - 1
    maximum = min(MAX_BINS, max(automatic, len(prepared.clipped_all), 1))
    return BinCountOptions(automatic=automatic, minimum=1, maximum=maximum)

plot_event_memory

plot_event_memory(source: dict[str, DataFrame] | str | Path | list[str | Path], *, column: str = 'rss_end_mb', labels: list[str] | None = None, baseline_label: str | None = None, show: str = 'both', bins: BinSpec = 'auto', bin_width: float | None = None, bin_origin: float | None = None, show_errors: bool = False, show_mean_lines: bool = True, alpha: float = 0.7, figsize: tuple[float, float] | None = None, outlier_threshold: float = 3.5, exclude_events: list[int] | None = None, palette: list[str] | None = None) -> Figure

Plot per-event memory (RSS) distributions for one or more runs.

Distribution panel shows a histogram of the chosen RSS, sampled at the end of each event. Sequence panel shows it vs event number.

Parameters:

Name	Type	Description	Default
source	dict[str, pd.DataFrame], str/Path, or list of str/Path	Pre-loaded dict from :func:~k4bench.analysis.loader.load_event_timing, a single log-dir path, or a list of log-dir paths.	required
column	str	Which RSS to draw, one of :data:EVENT_MEMORY_COLUMNS: total RSS ("rss_end_mb", the default), anonymous RSS ("rss_anon_end_mb") or file-backed RSS ("rss_file_end_mb"). Negative samples of the latter two, written where a /proc read failed, are dropped.	'rss_end_mb'
labels	list[str] or None	Restrict to these run labels.	None
baseline_label	str or None	Reference run for the ratio panel (multi-run only).	None
show	('both', 'distribution', 'sequence')	Which panels to display.	"both"
bins	int, str, or sequence of float	Bin count, a :func:numpy.histogram_bin_edges rule name, or explicit edges covering the pooled in-range data. Edges are resolved once, so all runs and the ratio panel share exactly the same bins.	'auto'
bin_width	float or None	Fixed bin width in data units. Mutually exclusive with bins. The width and origin define a stable grid across separate figures, whereas bins="auto" derives each figure's binning from its pooled data.	None
bin_origin	float or None	Origin of the fixed-width grid. Used only with bin_width and defaults to zero.	None
show_errors	bool	Draw Poisson √N uncertainties on the bin contents.	False
show_mean_lines	bool	Draw the dashed per-run mean line on the distribution panel.	True
alpha	float	Opacity of the filled bars (default: 0.7). The step outline drawn over each histogram stays fully opaque, so lowering this fades overlapping fills without losing any distribution; alpha=0 leaves outlines only.	0.7
figsize	(width, height) or None	Figure size in inches (converted to pixels at 96 dpi).	None
outlier_threshold	float	MAD-based modified Z-score threshold for x-range clipping.	3.5
exclude_events	list[int] or None	Event numbers to exclude. Defaults to [0].	None

Returns:

Type	Description
Figure

Source code in k4bench/analysis/plots/event.py

def plot_event_memory(
    source: dict[str, pd.DataFrame] | str | Path | list[str | Path],
    *,
    column: str = "rss_end_mb",
    labels: list[str] | None = None,
    baseline_label: str | None = None,
    show: str = "both",
    bins: BinSpec = "auto",
    bin_width: float | None = None,
    bin_origin: float | None = None,
    show_errors: bool = False,
    show_mean_lines: bool = True,
    alpha: float = 0.7,
    figsize: tuple[float, float] | None = None,
    outlier_threshold: float = 3.5,
    exclude_events: list[int] | None = None,
    palette: list[str] | None = None,
) -> go.Figure:
    """Plot per-event memory (RSS) distributions for one or more runs.

    Distribution panel shows a histogram of the chosen RSS, sampled at the end
    of each event. Sequence panel shows it vs event number.

    Parameters
    ----------
    source : dict[str, pd.DataFrame], str/Path, or list of str/Path
        Pre-loaded dict from :func:`~k4bench.analysis.loader.load_event_timing`,
        a single log-dir path, or a list of log-dir paths.
    column : str
        Which RSS to draw, one of :data:`EVENT_MEMORY_COLUMNS`: total RSS
        (``"rss_end_mb"``, the default), anonymous RSS (``"rss_anon_end_mb"``)
        or file-backed RSS (``"rss_file_end_mb"``). Negative samples of the
        latter two, written where a /proc read failed, are dropped.
    labels : list[str] or None
        Restrict to these run labels.
    baseline_label : str or None
        Reference run for the ratio panel (multi-run only).
    show : {"both", "distribution", "sequence"}
        Which panels to display.
    bins : int, str, or sequence of float
        Bin count, a :func:`numpy.histogram_bin_edges` rule name, or explicit
        edges covering the pooled in-range data. Edges are resolved once, so all
        runs and the ratio panel share exactly the same bins.
    bin_width : float or None
        Fixed bin width in data units. Mutually exclusive with ``bins``. The
        width and origin define a stable grid across separate figures, whereas
        ``bins="auto"`` derives each figure's binning from its pooled data.
    bin_origin : float or None
        Origin of the fixed-width grid. Used only with ``bin_width`` and defaults
        to zero.
    show_errors : bool
        Draw Poisson ``√N`` uncertainties on the bin contents.
    show_mean_lines : bool
        Draw the dashed per-run mean line on the distribution panel.
    alpha : float
        Opacity of the filled bars (default: 0.7).  The step outline drawn over
        each histogram stays fully opaque, so lowering this fades overlapping
        fills without losing any distribution; ``alpha=0`` leaves outlines only.
    figsize : (width, height) or None
        Figure size in inches (converted to pixels at 96 dpi).
    outlier_threshold : float
        MAD-based modified Z-score threshold for x-range clipping.
    exclude_events : list[int] or None
        Event numbers to exclude.  Defaults to ``[0]``.

    Returns
    -------
    plotly.graph_objects.Figure
    """
    if column not in EVENT_MEMORY_COLUMNS:
        raise ValueError(
            f"column must be one of {sorted(EVENT_MEMORY_COLUMNS)}, got {column!r}."
        )
    xlabel, yseq_label, stat_prefix = EVENT_MEMORY_COLUMNS[column]
    return _plot_event_metric(
        source,
        column=column,
        xlabel=xlabel,
        yseq_label=yseq_label,
        stat_prefix=stat_prefix,
        stat_unit="MB",
        warn_name="plot_event_memory",
        warn_unit="MB",
        labels=labels,
        baseline_label=baseline_label,
        show=show,
        bins=bins,
        bin_width=bin_width,
        bin_origin=bin_origin,
        show_errors=show_errors,
        show_mean_lines=show_mean_lines,
        alpha=alpha,
        figsize=figsize,
        outlier_threshold=outlier_threshold,
        exclude_events=exclude_events,
        palette=palette,
    )

plot_event_timing

plot_event_timing(source: dict[str, DataFrame] | str | Path | list[str | Path], *, labels: list[str] | None = None, baseline_label: str | None = None, show: str = 'both', bins: BinSpec = 'auto', bin_width: float | None = None, bin_origin: float | None = None, show_errors: bool = False, show_mean_lines: bool = True, alpha: float = 0.7, figsize: tuple[float, float] | None = None, outlier_threshold: float = 3.5, exclude_events: list[int] | None = None, palette: list[str] | None = None, quantity: str = 'Event time', references: Mapping[str, str] | None = None) -> Figure

Plot per-event timing distributions for one or more runs.

Single run: histogram with μ ± SEM and σ ± SE(σ) shown as an annotation. Multiple runs: overlaid histograms and, with show="both", bin-by-bin and per-event ratio panels for every non-baseline run against the reference.

Parameters:

Name	Type	Description	Default
source	dict[str, pd.DataFrame], str/Path, or list of str/Path	Pre-loaded dict from :func:~k4bench.analysis.loader.load_event_timing, a single log-dir path, or a list of log-dir paths.	required
labels	list[str] or None	Restrict to these run labels.	None
baseline_label	str or None	Reference run for the ratio panel (multi-run only).	None
show	('both', 'distribution', 'sequence')	Which panels to display.	"both"
bins	int, str, or sequence of float	Bin count, a :func:numpy.histogram_bin_edges rule name, or explicit edges covering the pooled in-range data. Edges are resolved once, so all runs and the ratio panel share exactly the same bins.	'auto'
bin_width	float or None	Fixed bin width in data units. Mutually exclusive with bins. The width and origin define a stable grid across separate figures, whereas bins="auto" derives each figure's binning from its pooled data.	None
bin_origin	float or None	Origin of the fixed-width grid. Used only with bin_width and defaults to zero.	None
show_errors	bool	Draw Poisson √N uncertainties on the bin contents.	False
show_mean_lines	bool	Draw the dashed per-run mean line on the distribution panel.	True
alpha	float	Opacity of the filled bars (default: 0.7). The step outline drawn over each histogram stays fully opaque, so lowering this fades overlapping fills without losing any distribution; alpha=0 leaves outlines only.	0.7
figsize	(width, height) or None	Figure size in inches (converted to pixels at 96 dpi).	None
outlier_threshold	float	MAD-based modified Z-score threshold for x-range clipping.	3.5
exclude_events	list[int] or None	Event numbers to exclude. Defaults to [0].	None
quantity	str	What the times are, as the axes name them: an algorithm's own time in each event ("MyConformalTracking time"), say.	'Event time'
references	mapping or None	Each run's own reference for the ratio panels, in place of baseline_label: runs that time different things are each set against their own kind. A run that is its own reference, or has none, has no ratio; with no ratio at all the panels are left out.	None

Returns:

Type	Description
Figure

Source code in k4bench/analysis/plots/event.py

def plot_event_timing(
    source: dict[str, pd.DataFrame] | str | Path | list[str | Path],
    *,
    labels: list[str] | None = None,
    baseline_label: str | None = None,
    show: str = "both",
    bins: BinSpec = "auto",
    bin_width: float | None = None,
    bin_origin: float | None = None,
    show_errors: bool = False,
    show_mean_lines: bool = True,
    alpha: float = 0.7,
    figsize: tuple[float, float] | None = None,
    outlier_threshold: float = 3.5,
    exclude_events: list[int] | None = None,
    palette: list[str] | None = None,
    quantity: str = "Event time",
    references: Mapping[str, str] | None = None,
) -> go.Figure:
    """Plot per-event timing distributions for one or more runs.

    Single run: histogram with μ ± SEM and σ ± SE(σ) shown as an annotation.
    Multiple runs: overlaid histograms and, with ``show="both"``, bin-by-bin
    and per-event ratio panels for every non-baseline run against the reference.

    Parameters
    ----------
    source : dict[str, pd.DataFrame], str/Path, or list of str/Path
        Pre-loaded dict from :func:`~k4bench.analysis.loader.load_event_timing`,
        a single log-dir path, or a list of log-dir paths.
    labels : list[str] or None
        Restrict to these run labels.
    baseline_label : str or None
        Reference run for the ratio panel (multi-run only).
    show : {"both", "distribution", "sequence"}
        Which panels to display.
    bins : int, str, or sequence of float
        Bin count, a :func:`numpy.histogram_bin_edges` rule name, or explicit
        edges covering the pooled in-range data. Edges are resolved once, so all
        runs and the ratio panel share exactly the same bins.
    bin_width : float or None
        Fixed bin width in data units. Mutually exclusive with ``bins``. The
        width and origin define a stable grid across separate figures, whereas
        ``bins="auto"`` derives each figure's binning from its pooled data.
    bin_origin : float or None
        Origin of the fixed-width grid. Used only with ``bin_width`` and defaults
        to zero.
    show_errors : bool
        Draw Poisson ``√N`` uncertainties on the bin contents.
    show_mean_lines : bool
        Draw the dashed per-run mean line on the distribution panel.
    alpha : float
        Opacity of the filled bars (default: 0.7).  The step outline drawn over
        each histogram stays fully opaque, so lowering this fades overlapping
        fills without losing any distribution; ``alpha=0`` leaves outlines only.
    figsize : (width, height) or None
        Figure size in inches (converted to pixels at 96 dpi).
    outlier_threshold : float
        MAD-based modified Z-score threshold for x-range clipping.
    exclude_events : list[int] or None
        Event numbers to exclude.  Defaults to ``[0]``.
    quantity : str
        What the times are, as the axes name them: an algorithm's own time
        in each event (``"MyConformalTracking time"``), say.
    references : mapping or None
        Each run's own reference for the ratio panels, in place of
        ``baseline_label``: runs that time different things are each set
        against their own kind. A run that is its own reference, or has none,
        has no ratio; with no ratio at all the panels are left out.

    Returns
    -------
    plotly.graph_objects.Figure
    """
    return _plot_event_metric(
        source,
        column="event_time_s",
        xlabel=f"{quantity} (s)",
        yseq_label=f"{quantity} (s)",
        stat_prefix="μ",
        stat_unit="s",
        warn_name="plot_event_timing",
        warn_unit="s",
        labels=labels,
        baseline_label=baseline_label,
        show=show,
        bins=bins,
        bin_width=bin_width,
        bin_origin=bin_origin,
        show_errors=show_errors,
        show_mean_lines=show_mean_lines,
        alpha=alpha,
        figsize=figsize,
        outlier_threshold=outlier_threshold,
        exclude_events=exclude_events,
        palette=palette,
        references=references,
    )

plot_standing

plot_standing(values: DataFrame, name: str, measures: Sequence[Measure], *, color: str = ACCENT) -> Figure | None

A panel per one of measures that name has a value above zero in: every component of values (indexed by component) with one, as a column, the largest first, name's in color and pointed at with its value, the rest grey; over the panel, its rank and, for a measure whose values add up, its share of their sum, under its title and unit. A panel's axis is logarithmic unless its measure says otherwise. A panel's columns carry their component's name as customdata[0]. None when name has a value in none of them.

Source code in k4bench/analysis/plots/component.py

def plot_standing(
    values: pd.DataFrame, name: str, measures: Sequence[Measure], *, color: str = ACCENT,
) -> go.Figure | None:
    """A panel per one of *measures* that *name* has a value above zero in:
    every component of *values* (indexed by component) with one, as a column,
    the largest first, *name*'s in *color* and pointed at with its value, the
    rest grey; over the panel, its rank and, for a measure whose values add
    up, its share of their sum, under its title and unit. A panel's axis is
    logarithmic unless its measure says otherwise. A panel's columns carry their component's name as
    ``customdata[0]``. ``None`` when *name* has a value in none of them."""
    shown = []
    for measure in measures:
        column = values[measure.column]
        column = column[column.notna() & (column > 0)].sort_values(ascending=False, kind="stable")
        if name in column.index:
            shown.append((measure, column))
    if not shown:
        return None
    fig = make_subplots(rows=1, cols=len(shown), horizontal_spacing=0.07)
    for i, (measure, column) in enumerate(shown, start=1):
        place = column.index.get_loc(name)
        value = float(column[name])
        fig.add_trace(go.Bar(
            x=list(range(len(column))), y=column.to_numpy(),
            marker=dict(
                color=[color if n == name else MUTED_FILL for n in column.index],
                cornerradius=2, line=dict(width=0),
            ),
            customdata=[(n, measure.write(v)) for n, v in column.items()],
            hovertemplate="<b>%{customdata[0]}</b><br>%{customdata[1]}<extra></extra>",
            showlegend=False,
        ), row=1, col=i)
        axis = "" if i == 1 else str(i)
        # A log axis places annotations by the decade.
        fig.add_annotation(
            x=place, y=math.log10(value) if measure.log else value,
            xref=f"x{axis}", yref=f"y{axis}", text=measure.write(value),
            showarrow=True, arrowhead=0, arrowcolor=color, arrowwidth=1.5, ax=0, ay=-26,
            font=dict(color=_INK, size=11), bgcolor="rgba(255,255,255,0.85)",
        )
        rank = f"Rank {place + 1} of {len(column)}"
        if measure.total is not None and column.sum() > 0:
            share = value / column.sum()
            rank += f" · {'<1%' if share < 0.005 else f'{share:.0%}'} of {measure.total}"
        title = (f"<b>{measure.title}</b>"
                 f"<span style='color:{_MUTED}'>  {measure.unit}</span>")
        for text, shift, size, ink in ((title, 32, 13, _INK), (rank, 14, 12, _MUTED)):
            fig.add_annotation(
                text=text, x=0, xref=f"x{axis} domain", xanchor="left",
                y=1, yref="paper", yanchor="bottom", yshift=shift, showarrow=False,
                font=dict(color=ink, size=size),
            )
        fig.update_xaxes(visible=False, range=[-0.6, len(column) - 0.4], row=1, col=i)
        if measure.log:
            # Room above the largest for the arrow, a little below the least.
            low, high = math.log10(column.min()), math.log10(column.max())
            fig.update_yaxes(
                type="log", range=[low - 0.25, high + 0.9], dtick=1, row=1, col=i,
            )
        else:
            fig.update_yaxes(range=[0, float(column.max()) * 1.35], nticks=4, row=1, col=i)
        fig.update_yaxes(
            gridcolor=_GRID, color=_MUTED, tickformat=measure.tickformat,
            tickfont=dict(size=10), row=1, col=i,
        )
    fig.update_layout(
        template=_TEMPLATE, height=_HEIGHT, bargap=0.12,
        margin=dict(l=10, r=10, t=70, b=4),
    )
    return fig

plot_component_heap_events

plot_component_heap_events(start: Series, peak: Series, end: Series, *, before_first: float | None = None, after_finalize: float | None = None, color: str = _PALETTE[0], warm_up: int | None = 0) -> Figure | None

One component's heap event by event, in MB, each series indexed by event number: a bar from what it held as the event started (start) up to the most it held during it (peak), the event's working set, and a marker at what it held at the event's end (end). What it held before the first event and after finalize, where given, head the chart, so its axis spans the events alone. Without peaks (a job that ran events at once), the bars are left out and the marks drawn alone. The warm_up event is paler. None when there is nothing to draw.

Source code in k4bench/analysis/plots/heap.py

def plot_component_heap_events(
    start: pd.Series,
    peak: pd.Series,
    end: pd.Series,
    *,
    before_first: float | None = None,
    after_finalize: float | None = None,
    color: str = _PALETTE[0],
    warm_up: int | None = 0,
) -> go.Figure | None:
    """One component's heap event by event, in MB, each series indexed by
    event number: a bar from what it held as the event started (*start*) up to
    the most it held during it (*peak*), the event's working set, and a marker
    at what it held at the event's end (*end*). What it held before the first
    event and after finalize, where given, head the chart, so its axis spans
    the events alone. Without peaks (a job that ran events at once), the bars
    are left out and the marks drawn alone. The *warm_up* event is paler.
    ``None`` when there is nothing to draw.
    """
    if start.empty:
        return None
    events = start.sort_index().index
    start, peak, end = (s.reindex(events) for s in (start, peak, end))
    values = np.concatenate([start.to_numpy(), peak.to_numpy(), end.to_numpy()]).astype(float)
    values = values[np.isfinite(values)]
    if not len(values):
        return None
    low, high = float(values.min()), float(values.max())
    divisor, unit = _scale(high)
    opacity = [0.4 if e == warm_up else 1.0 for e in events]
    fig = go.Figure()
    fig.add_trace(go.Bar(
        x=events, y=(peak - start) / divisor, base=start / divisor,
        name="During the event (start to peak)",
        marker=dict(color=color, opacity=opacity, line=dict(width=0)),
        customdata=np.column_stack([start / divisor, peak / divisor, (peak - start) / divisor]),
        hovertemplate=(
            f"start %{{customdata[0]:.3g}} {unit} · peak %{{customdata[1]:.3g}} {unit}"
            f" · working set %{{customdata[2]:.3g}} {unit}<extra></extra>"
        ),
    ))
    fig.add_trace(go.Scatter(
        x=events, y=end / divisor, name="At the event's end", mode="markers",
        marker=dict(color=_INK, size=6, symbol="line-ew-open", line=dict(width=2)),
        hovertemplate=f"end %{{y:.3g}} {unit}<extra></extra>",
    ))
    moments = " · ".join(
        f"{text}: {format_mb(value)}"
        for value, text in ((before_first, INITIALIZED), (after_finalize, FINALIZED))
        if value is not None and math.isfinite(value)
    )
    pad = max((high - low) * 0.08, abs(high) * 0.002, 1e-6)
    fig.update_layout(
        template=_TEMPLATE, height=380, margin=dict(l=20, r=20, t=40 if moments else 20, b=20),
        title=dict(text=moments, x=0, xanchor="left", font=dict(size=13, color=_SECONDARY_INK))
        if moments else None,
        hovermode="x unified", bargap=0.1, legend=bottom_legend(),
    )
    fig.update_xaxes(title_text="Event number")
    fig.update_yaxes(title_text=f"Heap held ({unit})", range=[(low - pad) / divisor, (high + pad) / divisor])
    _style(fig)
    return fig

plot_heap_timeline

plot_heap_timeline(held: DataFrame, before_first: Series, after_finalize: Series, named: list[str], *, colors: Mapping[str, str] | None = None, rss: Series | None = None, change: bool = False, warm_up: int | None = 0, steps: DataFrame | None = None, log_y: bool = False) -> Figure | None

The heap each series held, from initialization through each event to finalization.

held is events × series, in MB at each event's end, indexed by event number: the job's components (execute["held_bytes"]) or groups of them, :data:~k4bench.analysis.heap.NO_COMPONENT among them. before_first and after_finalize are the same series' heap before the first event and after the job's finalize.

Stacked, the named series are a band each, the most stable lowest; the rest are one "Other (k)" band, and what no component allocated the lowest band of all, both in neutral greys. The bands are opaque and set apart by a gap in the surface colour; one wide enough at the last event carries its name inside it there. A dashed line is rss, the anonymous RSS of the job at each event's end (MB, indexed by event number) where given; the top of the stack, the heap of the whole job, is in the hover. With change, each of the named series is instead a line of what it held at each event's end beyond what it held at the first event's end, so growth rises and a step jumps, and the rest and what no component allocated are two grey lines: together the lines make the whole heap's change, so growth in a series not named still shows. colors maps a series to its colour; the others take the default palette's colours but its grey, in order, which a caller drawing more series than it has colours must not rely on. The warm_up event is shaded. steps marks the events where a series rose at once, one row each with its series, event_number and step_mb (:meth:~k4bench.analysis.heap.ComponentHeap.steps): a marker in the series' colour on the top of the stack, one for all of an event's steps (in the largest's colour, every one in the hover), or on the series' line with change (a step of a series not named is not marked there). With log_y, the stack is drawn on a log axis reaching below the lowest band, so the small bands under a large one stay readable; it does not apply with change, whose lines cross zero. None when there is nothing to draw.

Source code in k4bench/analysis/plots/heap.py

def plot_heap_timeline(
    held: pd.DataFrame,
    before_first: pd.Series,
    after_finalize: pd.Series,
    named: list[str],
    *,
    colors: Mapping[str, str] | None = None,
    rss: pd.Series | None = None,
    change: bool = False,
    warm_up: int | None = 0,
    steps: pd.DataFrame | None = None,
    log_y: bool = False,
) -> go.Figure | None:
    """The heap each series held, from initialization through each event to
    finalization.

    *held* is events × series, in MB at each event's end, indexed by event
    number: the job's components (``execute["held_bytes"]``) or groups of them,
    :data:`~k4bench.analysis.heap.NO_COMPONENT` among them.
    *before_first* and *after_finalize* are the same series' heap before the
    first event and after the job's finalize.

    Stacked, the *named* series are a band each, the most stable lowest; the
    rest are one "Other (k)" band, and what no component allocated the lowest
    band of all, both in neutral greys. The bands are opaque and set apart by
    a gap in the surface colour; one wide enough at the last event carries its
    name inside it there. A dashed line is *rss*, the anonymous RSS of the job
    at each event's end (MB, indexed by event number) where given; the top of
    the stack, the heap of the whole job, is in the hover. With *change*, each
    of the *named* series is instead a line of what it held at each event's
    end beyond what it held at the first event's end, so growth rises and a
    step jumps, and the rest and what no component allocated are two grey
    lines: together the lines make the whole heap's change, so growth in a
    series not named still shows. *colors* maps a series to its colour; the others take the
    default palette's colours but its grey, in order, which a caller drawing
    more series than it has colours must not rely on. The
    *warm_up* event is shaded. *steps* marks the events where a series
    rose at once, one row each with its ``series``, ``event_number`` and
    ``step_mb`` (:meth:`~k4bench.analysis.heap.ComponentHeap.steps`): a
    marker in the series' colour on the top of the stack, one for all of an
    event's steps (in the largest's colour, every one in the hover), or on
    the series' line with *change* (a step of a series not named is not
    marked there). With *log_y*, the stack is drawn on a log axis reaching
    below the lowest band, so the small bands under a large one stay
    readable; it does not apply with *change*, whose lines cross zero.
    ``None`` when there is nothing to draw.
    """
    if held.empty:
        return None
    held = held.sort_index()
    events = [int(e) for e in held.index]
    colors = colors or {}
    vivid = _vivid(_PALETTE)
    palette = iter(vivid * (len(named) // len(vivid) + 1))
    color_of = {name: colors.get(name) or next(palette) for name in named}
    fig = go.Figure()
    log_range = None

    if change:
        grown = held - held.iloc[0]
        lines = [name for name in named if name in grown.columns and name != NO_COMPONENT]
        rest = [c for c in grown.columns if c not in lines and c != NO_COMPONENT]
        series = [(name, grown[name], dict(color=color_of[name], width=2)) for name in lines]
        if rest:
            series.append((f"Other ({len(rest)})", grown[rest].sum(axis=1), dict(color=_OTHER_LINE, width=2)))
        if NO_COMPONENT in grown.columns:
            series.append((NO_COMPONENT_LABEL, grown[NO_COMPONENT], dict(color=_NO_COMPONENT_LINE, width=2, dash="dot")))
        divisor, unit = _scale(max((float(values.abs().max()) for _, values, _ in series), default=0.0))
        for label, values, line in series:
            fig.add_trace(go.Scatter(
                x=[str(e) for e in events], y=values / divisor, name=label, mode="lines", line=line,
                hovertemplate=f"{label}: %{{y:+.3g}} {unit}<extra></extra>",
            ))
        fig.add_hline(y=0, line=dict(color=_RULE, width=1))
        moments = [str(e) for e in events]
        y_title = f"Change since the first event ({unit})"
    else:
        moments = [INITIALIZED, *(str(e) for e in events), FINALIZED]
        table = pd.concat([
            before_first.reindex(held.columns).to_frame(INITIALIZED).T,
            held.set_axis([str(e) for e in events]),
            after_finalize.reindex(held.columns).to_frame(FINALIZED).T,
        ]).fillna(0.0)
        bands = _named_bands(held, named)
        rest = [c for c in held.columns if c not in bands and c != NO_COMPONENT]
        stack: list[tuple[str, pd.Series, str]] = []
        if NO_COMPONENT in table.columns:
            stack.append((NO_COMPONENT_LABEL, table[NO_COMPONENT], _NO_COMPONENT_FILL))
        if rest:
            stack.append((f"Other ({len(rest)})", table[rest].sum(axis=1), _OTHER_FILL))
        stack += [(name, table[name], color_of[name]) for name in bands]
        total = table.sum(axis=1)
        largest = max(float(total.max()), float(rss.max()) if rss is not None and len(rss) else 0.0)
        divisor, unit = _scale(largest)
        if log_y:
            log_range = _log_range(stack, rss)
        last, below = str(events[-1]), 0.0
        for label, values, color in stack:
            fig.add_trace(go.Scatter(
                x=moments, y=values / divisor, name=label, mode="lines", stackgroup="heap",
                line=dict(color=_SURFACE, width=1.5), fillcolor=color,
                hovertemplate=f"{label}: %{{y:.3g}} {unit}<extra></extra>",
            ))
            width = float(values[last])
            if log_range is None:
                share, y = (width / largest if largest else 0.0), (below + width / 2) / divisor
            else:
                # What a band spans on the axis, and its middle, are in decades.
                bottom, top = log_range
                low = max(math.log10(below), bottom) if below > 0 else bottom
                high = math.log10(below + width) if below + width > 0 else bottom
                share = (high - low) / (top - bottom)
                y = (low + high) / 2 - math.log10(divisor)
            if share >= _LABEL_SHARE:
                fig.add_annotation(
                    x=last, y=y, text=label, showarrow=False,
                    xanchor="right", xshift=-8, font=dict(size=11, color=_text_on(color)),
                )
            below += width
        fig.add_trace(go.Scatter(
            x=moments, y=total / divisor, mode="lines", line=dict(color="rgba(0,0,0,0)"),
            showlegend=False, hovertemplate=f"<b>Heap held: %{{y:.3g}} {unit}</b><extra></extra>",
        ))
        if rss is not None and len(rss):
            rss = rss.reindex(events)
            fig.add_trace(go.Scatter(
                x=[str(e) for e in events], y=rss / divisor, name="Anonymous RSS (timed run)",
                mode="lines", line=dict(color=_INK, width=1.5, dash="dash"),
                hovertemplate=f"Anonymous RSS (timed run): %{{y:.3g}} {unit}<extra></extra>",
            ))
        # The two moments are not events: a hairline sets each apart.
        for x in (0.5, len(moments) - 1.5):
            fig.add_vline(x=x, line=dict(color=_RULE, width=1))
        y_title = f"Heap held ({unit})"

    if steps is not None and len(steps):
        marks = steps[steps["event_number"].isin(events)]
        if change:
            marks = marks[marks["series"].isin(lines)]
        marks = marks.assign(text=[
            f"{NO_COMPONENT_LABEL if name == NO_COMPONENT else name} {format_mb(mb, signed=True)}"
            for name, mb in zip(marks["series"], marks["step_mb"])
        ])
        if change:
            heights = [grown.at[e, name] for name, e in zip(marks["series"], marks["event_number"])]
        elif len(marks):
            # The steps of one event share the top of the stack: one marker
            # names them all, in the colour of the largest.
            marks = (
                marks.sort_values("step_mb", ascending=False, kind="stable")
                .groupby("event_number", sort=True, as_index=False)
                .agg(series=("series", "first"), text=("text", ", ".join))
            )
            heights = [total[str(e)] for e in marks["event_number"]]
        if len(marks):
            fig.add_trace(go.Scatter(
                x=[str(e) for e in marks["event_number"]], y=np.asarray(heights, dtype=float) / divisor,
                name="Step",
                mode="markers",
                marker=dict(
                    symbol="triangle-down", size=11, line=dict(color=_SURFACE, width=1),
                    color=[color_of.get(name, _OTHER_LINE) for name in marks["series"]],
                ),
                text=marks["text"].tolist(),
                hovertemplate="Step: %{text}<extra></extra>",
            ))

    if warm_up is not None and warm_up in events:
        at = moments.index(str(warm_up))
        fig.add_vrect(
            x0=at - 0.5, x1=at + 0.5, fillcolor="rgba(0,0,0,0.05)", line_width=0, layer="below",
            annotation_text="warm-up", annotation_position="top left",
            annotation_font=dict(size=10, color=_MUTED),
        )
    ticks = _event_ticks(events, clear_ends=not change)
    if not change:
        ticks = [INITIALIZED, *ticks, FINALIZED]
    fig.update_layout(
        template=_TEMPLATE, height=480, margin=dict(l=20, r=20, t=30, b=20),
        hovermode="x unified", legend=bottom_legend(),
    )
    fig.update_xaxes(
        type="category", tickmode="array", tickvals=ticks, tickangle=0, title_text="Event number",
    )
    if log_range is not None:
        bottom, top = log_range
        # Ticks at 1, 2 and 5 of each decade, not every digit, each written
        # in full (0.5, not 5).
        fig.update_yaxes(
            title_text=y_title, type="log", dtick="D2", tickformat=",~g",
            range=[bottom - math.log10(divisor), top - math.log10(divisor)],
        )
    else:
        fig.update_yaxes(title_text=y_title, rangemode="tozero" if not change else "normal")
    _style(fig)
    return fig

plot_run_overview

plot_run_overview(results: DataFrame | str | Path | list[str | Path], *, labels: list[str] | None = None, metrics: list[tuple[str, str]] | None = None, relative: bool = True, baseline_label: str | None = None) -> Figure

Plot run metrics for all runs in a 2 × 2 panel grid.

Each metric is drawn as a horizontal bar chart with value annotations.

Parameters:

Name	Type	Description	Default
results	pd.DataFrame, str/Path, or list of str/Path	Results DataFrame, a single log-dir path, or a list of log-dir paths for multi-detector comparisons.	required
labels	list[str] or None	Show only these run labels.	None
metrics	list of (column, axis-label) pairs or None	Which metrics to plot. Defaults to wall_time_s, peak_rss_mb, user_cpu_s, events_per_sec.	None
relative	bool	If True, normalise every metric to the baseline run (= 100 %).	True
baseline_label	str or None	Which run to treat as 100 % when relative=True.	None

Returns:

Type	Description
Figure

Source code in k4bench/analysis/plots/overview.py

def plot_run_overview(
    results: pd.DataFrame | str | Path | list[str | Path],
    *,
    labels: list[str] | None = None,
    metrics: list[tuple[str, str]] | None = None,
    relative: bool = True,
    baseline_label: str | None = None,
) -> go.Figure:
    """Plot run metrics for all runs in a 2 × 2 panel grid.

    Each metric is drawn as a horizontal bar chart with value annotations.

    Parameters
    ----------
    results : pd.DataFrame, str/Path, or list of str/Path
        Results DataFrame, a single log-dir path, or a list of log-dir paths
        for multi-detector comparisons.
    labels : list[str] or None
        Show only these run labels.
    metrics : list of (column, axis-label) pairs or None
        Which metrics to plot.  Defaults to wall_time_s, peak_rss_mb,
        user_cpu_s, events_per_sec.
    relative : bool
        If ``True``, normalise every metric to the baseline run (= 100 %).
    baseline_label : str or None
        Which run to treat as 100 % when ``relative=True``.

    Returns
    -------
    plotly.graph_objects.Figure
    """
    det_title = _detector_title(results)
    results = _ensure_df(results)

    if metrics is None:
        metrics = _OVERVIEW_METRICS

    if labels is not None:
        results = results[results["label"].apply(
            lambda lbl: lbl in labels or any(lbl.endswith(f"/{w}") for w in labels)
        )]

    metric_cols = [col for col, _ in metrics if col in results.columns]
    df = results.dropna(subset=metric_cols, how="all").copy()

    # Capture label order before the wall-time sort so that baseline resolution
    # is deterministic (first loaded row) regardless of display order.
    load_order_labels = df["label"].tolist()

    baseline_vals: dict[str, float] = {}
    if relative:
        _bl = baseline_label if baseline_label is not None else _default_baseline(load_order_labels)
        bl_mask = df["label"].apply(lambda lbl: _matches_baseline(lbl, _bl))
        if not bl_mask.any():
            hint = " Pass baseline_label=... to specify the reference run." if baseline_label is None else ""
            raise ValueError(f"baseline_label '{_bl}' not found for relative=True.{hint}")
        if bl_mask.sum() > 1:
            matched = df.loc[bl_mask, "label"].tolist()
            raise ValueError(
                f"baseline_label '{_bl}' matches multiple runs: {matched}. "
                "Pass the full prefixed label to disambiguate."
            )
        for col, _ in metrics:
            if col in df.columns:
                bv = float(df.loc[bl_mask, col].iloc[0])
                baseline_vals[col] = bv
                if bv == 0:
                    raise ValueError(
                        f"Baseline value for metric '{col}' is 0 — cannot normalise to percentage."
                    )
                df[col] = df[col] / bv * 100

    if "wall_time_s" in df.columns:
        df = df.sort_values("wall_time_s", ascending=True)

    run_labels = df["label"].tolist()
    n_runs = len(run_labels)


    n_metrics = len(metrics)
    if n_metrics == 0:
        raise ValueError("metrics must contain at least one metric")

    ncols = 2
    nrows = (n_metrics + 1) // ncols

    subplot_titles = []
    for col, ylabel in metrics:
        if relative:
            subplot_titles.append(f"{ylabel.split(' (')[0]} %")
        else:
            subplot_titles.append(ylabel)

    # Keep the inter-row gap fixed at ~80 px regardless of figure height.
    fig_height = max(300, 45 * n_runs + 150) * nrows
    v_spacing = (80 / fig_height) if nrows > 1 else 0.0

    fig = make_subplots(
        rows=nrows, cols=ncols,
        subplot_titles=subplot_titles,
        shared_yaxes=True,
        horizontal_spacing=0.12,
        vertical_spacing=v_spacing,
    )

    for idx, (col, _ylabel) in enumerate(metrics):
        row = idx // ncols + 1
        col_num = idx % ncols + 1

        if col not in df.columns:
            continue

        values = df[col].tolist()
        valid_v = [v for v in values if pd.notna(v)]
        x_max = max(valid_v) if valid_v else 1.0

        # Gradient: green (best) → yellow → red (worst), ranked per metric.
        col_series = df.set_index("label")[col]
        n_valid = int(col_series.notna().sum())
        lower_is_better = METRICS[col].lower_is_better if col in METRICS else False
        ranks = col_series.rank(ascending=bool(lower_is_better), na_option="keep")
        denom = max(n_valid - 1, 1)
        sample_pts = []
        for lbl in run_labels:
            if lbl not in ranks.index or pd.isna(ranks.at[lbl]):
                sample_pts.append(0.0)
            else:
                sample_pts.append(float(1.0 - (float(ranks.at[lbl]) - 1.0) / denom))
        bar_colors = pc.sample_colorscale("RdYlGn", sample_pts)

        for i, (run_label, val) in enumerate(zip(run_labels, values)):
            if relative:
                text = f"{val:.1f}%" if pd.notna(val) else ""
                hover_tmpl = "%{y}<br><b>%{x:.1f}%</b><extra></extra>"
            else:
                text = f"{val:.4g}" if pd.notna(val) else ""
                hover_tmpl = "%{y}<br><b>%{x:.4g}</b><extra></extra>"

            fig.add_trace(
                go.Bar(
                    x=[val],
                    y=[run_label],
                    orientation="h",
                    marker_color=bar_colors[i],
                    marker_line_width=0,
                    name=run_label,
                    showlegend=False,
                    text=[text],
                    textposition="outside",
                    textfont=dict(size=9, color="#444444"),
                    hovertemplate=hover_tmpl,
                ),
                row=row, col=col_num,
            )

        axis_kw = dict(range=[0, x_max * 1.22], row=row, col=col_num)
        if relative:
            axis_kw["ticksuffix"] = "%"
            fig.add_vline(
                x=100, line_dash="dash", line_color="black",
                line_width=0.8, opacity=0.4, row=row, col=col_num,
            )
        fig.update_xaxes(**axis_kw)

    title = f"Run Metrics Overview — {det_title}" if det_title else "Run Metrics Overview"
    fig.update_layout(
        title_text=title,
        title_font=dict(size=16, color="#222222"),
        template=_TEMPLATE,
        height=fig_height,
        bargap=0.35,
        showlegend=False,
        margin=dict(l=20, r=20, t=80, b=40),
    )

    return fig

plot_event_composition

plot_event_composition(times: DataFrame, components: list[str], *, colors: Mapping[str, str] | None = None, sort: bool = False, warm_up: int | None = 0) -> Figure

The event loop event by event: a column per event stacking each of components' self time in it, the rest as "Other".

times is the job's execute["wall_s"] (events × components, NaN where a component did not run). colors maps a component to its colour, as :func:plot_phase_component_pie takes it. sort orders the events by their time, longest first, instead of by event number. The warm_up event is drawn paler. A component's bars carry its name as their customdata; "Other"'s carry none.

Source code in k4bench/analysis/plots/phases.py

def plot_event_composition(
    times: pd.DataFrame,
    components: list[str],
    *,
    colors: Mapping[str, str] | None = None,
    sort: bool = False,
    warm_up: int | None = 0,
) -> go.Figure:
    """The event loop event by event: a column per event stacking each of
    *components*' self time in it, the rest as "Other".

    *times* is the job's ``execute["wall_s"]`` (events × components, ``NaN``
    where a component did not run). *colors* maps a component to its colour,
    as :func:`plot_phase_component_pie` takes it. *sort* orders the events by
    their time, longest first, instead of by event number. The *warm_up*
    event is drawn paler. A component's bars carry its name as their
    ``customdata``; "Other"'s carry none.
    """
    times = times.fillna(0.0)
    named = [c for c in components if c in times.columns]
    rest = [c for c in times.columns if c not in named]
    series = {c: times[c] for c in named}
    if rest and (times[rest].to_numpy() > 0).any():
        series[f"Other ({len(rest)})"] = times[rest].sum(axis=1)
    totals = times.sum(axis=1)
    order = totals.sort_values(ascending=False).index if sort else times.index
    labels = [str(event) for event in order]
    opacity = [0.45 if event == warm_up else 1.0 for event in order]

    palette = iter(_PALETTE * (len(named) // len(_PALETTE) + 1))
    fig = go.Figure()
    for name, seconds in series.items():
        color = (
            _OTHER_COLOR if name not in named
            else (colors or {}).get(name) or next(palette)
        )
        fig.add_trace(go.Bar(
            x=labels, y=seconds.reindex(order), name=name,
            customdata=[(name,)] * len(labels) if name in named else None,
            marker=dict(color=color, opacity=opacity),
            hovertemplate=f"{name}: %{{y:.3g}} s<extra></extra>",
        ))
    fig.update_layout(
        template=_TEMPLATE,
        barmode="stack",
        bargap=0.15,
        height=460,
        margin=dict(l=20, r=20, t=20, b=20),
        hovermode="x unified",
        legend=bottom_legend(),
    )
    fig.update_xaxes(
        type="category",
        title_text="Event, longest first" if sort else "Event number",
        showticklabels=not sort,
    )
    fig.update_yaxes(title_text="Self time (s)")
    return fig

plot_phase_component_pie

plot_phase_component_pie(seconds: Series, *, top_n: int = 5, colors: Mapping[str, str] | None = None, title: str | None = None, duration: float | None = None, unit: str = 's') -> Figure

A donut of one phase's time by component: the :func:top_components and the rest as one "Other (k)" slice, the total in the middle.

seconds is component → self seconds in the phase (:meth:~k4bench.analysis.components.ComponentTiming.phase_component_seconds). duration is the phase's elapsed seconds (:meth:~k4bench.analysis.components.ComponentTiming.phase_seconds): the part of it no component was charged for is one more slice, :data:UNATTRIBUTED, and the middle is the phase's duration. Without it, the middle is the components' summed time; when that exceeds duration (components run in parallel), too, marked as component time. colors maps a component to its colour, so a component that several donuts show keeps one; the others take the palette in order. Parts of another quantity than time give its unit.

Source code in k4bench/analysis/plots/phases.py

def plot_phase_component_pie(
    seconds: pd.Series,
    *,
    top_n: int = 5,
    colors: Mapping[str, str] | None = None,
    title: str | None = None,
    duration: float | None = None,
    unit: str = "s",
) -> go.Figure:
    """A donut of one phase's time by component: the :func:`top_components`
    and the rest as one "Other (k)" slice, the total in the middle.

    *seconds* is component → self seconds in the phase
    (:meth:`~k4bench.analysis.components.ComponentTiming.phase_component_seconds`).
    *duration* is the phase's elapsed seconds
    (:meth:`~k4bench.analysis.components.ComponentTiming.phase_seconds`):
    the part of it no component was charged for is one more slice,
    :data:`UNATTRIBUTED`, and the middle is the phase's duration. Without
    it, the middle is the components' summed time; when that exceeds
    *duration* (components run in parallel), too, marked as component time.
    *colors* maps a component to its colour, so a component that several
    donuts show keeps one; the others take the palette in order. Parts of
    another quantity than time give its *unit*.
    """
    seconds = seconds.dropna()
    seconds = seconds[seconds > 0].sort_values(ascending=False)
    measured = float(seconds.sum())
    top = top_components(seconds, top_n)
    rest = seconds.drop(index=top)
    names = list(top)
    values = [float(seconds[name]) for name in top]
    palette = iter(_PALETTE * (len(top) // len(_PALETTE) + 1))
    slice_colors = [(colors or {}).get(name) or next(palette) for name in top]
    if len(rest):
        names.append(f"Other ({len(rest)})")
        values.append(float(rest.sum()))
        slice_colors.append(_OTHER_COLOR)
    if duration is not None and duration > measured:
        names.append(UNATTRIBUTED)
        values.append(duration - measured)
        slice_colors.append(_UNATTRIBUTED_COLOR)
    total = float(sum(values))
    shares = [v / total if total else 0.0 for v in values]
    centre = f"<b>{_total(total, unit)}</b>"
    if duration is not None and measured > duration:
        centre += "<br>component time"

    fig = go.Figure(go.Pie(
        labels=names, values=values, hole=0.58, sort=False, direction="clockwise",
        marker=dict(colors=slice_colors, line=dict(color="white", width=2)),
        text=[f"{share:.0%}" if share >= _MIN_LABEL_SHARE else "" for share in shares],
        textinfo="text", textposition="inside", insidetextorientation="horizontal",
        customdata=[_amount(v, unit) for v in values],
        hovertemplate="<b>%{label}</b><br>%{customdata} · %{percent}<extra></extra>",
    ))
    fig.add_annotation(
        text=centre, x=0.5, y=0.5, xref="paper", yref="paper",
        showarrow=False, font=dict(size=18),
    )
    fig.update_layout(
        template=_TEMPLATE,
        title=dict(text=title, x=0.5, xanchor="center") if title else None,
        height=440,
        margin=dict(l=10, r=10, t=50 if title else 10, b=10),
        legend=dict(orientation="v", x=0.5, xanchor="center", y=-0.05, yanchor="top"),
    )
    return fig

plot_phase_history

plot_phase_history(series: DataFrame, *, palette: list[str] = _PALETTE, relative: bool = False, value: str = 'seconds', unit: str = 's', y_title: str = 'Time (s)', whole: str = 'phase', all_runs: bool = False) -> Figure | None

One phase's time per nightly tag, a column per tag stacking :func:phase_component_series' series in their order, "Other" grey and :data:UNATTRIBUTED paler: a column's height is the series' total, written above it, the phase's duration when the series carry its unattributed part. relative draws each series as its share of the column. A tag run more than once shows its latest run (:func:latest_run_per_tag). None when there is nothing to draw.

Series of another quantity than time (their :func:phase_component_series value) give its column as value, its unit, the y_title of the axis and what a column is the whole of.

all_runs draws a column per run instead, every run of a tag, in the order they were measured: each named by the night it ran, the hover adding its release (:func:run_label). Two runs are never one column, whatever night or tag they share.

Source code in k4bench/analysis/plots/phases.py

def plot_phase_history(
    series: pd.DataFrame,
    *,
    palette: list[str] = _PALETTE,
    relative: bool = False,
    value: str = "seconds",
    unit: str = "s",
    y_title: str = "Time (s)",
    whole: str = "phase",
    all_runs: bool = False,
) -> go.Figure | None:
    """One phase's time per nightly tag, a column per tag stacking
    :func:`phase_component_series`' series in their order, "Other" grey and
    :data:`UNATTRIBUTED` paler: a column's height is the series' total,
    written above it, the phase's duration when the series carry its
    unattributed part. *relative* draws each series as its share of the
    column. A tag run more than once shows its latest run
    (:func:`latest_run_per_tag`). ``None`` when there is nothing to draw.

    Series of another quantity than time (their :func:`phase_component_series`
    *value*) give its column as *value*, its *unit*, the *y_title* of the
    axis and what a column is the *whole* of.

    *all_runs* draws a column per run instead, every run of a tag, in the
    order they were measured: each named by the night it ran, the hover
    adding its release (:func:`run_label`). Two runs are never one column,
    whatever night or tag they share.
    """
    if series.empty:
        return None
    # The series' order is the input's: longest first, "Other" and
    # "Unattributed" last.
    names = list(dict.fromkeys(series["label"]))
    series = history_rows(series.dropna(subset=["run_date"]), all_runs=all_runs)
    if all_runs:
        runs = (
            series[["x_date", "run_id", "run_date", "k4h_release"]]
            .drop_duplicates(["x_date", "run_id"])
            .sort_values(["run_date", "x_date", "run_id"], kind="stable")
        )
        index = pd.MultiIndex.from_frame(runs[["x_date", "run_id"]])
        table = series.pivot_table(
            index=["x_date", "run_id"], columns="label", values=value, aggfunc="sum",
            fill_value=0.0,
        ).reindex(index=index, columns=names, fill_value=0.0)
        dates = [run_label(d, r) for d, r in zip(runs["run_date"], runs["k4h_release"])]
        table.index = dates
    else:
        dates = sorted(pd.to_datetime(series["x_date"]).unique())
        table = series.pivot_table(
            index="x_date", columns="label", values=value, aggfunc="sum", fill_value=0.0,
        ).reindex(index=dates, columns=names, fill_value=0.0)
    totals = table.sum(axis=1)

    fig = go.Figure()
    for i, name in enumerate(names):
        seconds = table[name]
        share = (seconds / totals.where(totals > 0)).fillna(0.0)
        color = (
            _OTHER_COLOR if name.startswith("Other (")
            else _UNATTRIBUTED_COLOR if name == UNATTRIBUTED
            else palette[i % len(palette)]
        )
        fig.add_trace(go.Bar(
            x=dates, y=share * 100 if relative else seconds, name=name,
            marker=dict(
                color=color,
                line=dict(color="white", width=1),
            ),
            customdata=list(zip(seconds, share)),
            hovertemplate=(
                f"{name}: %{{customdata[0]:.3g}} {unit} · %{{customdata[1]:.0%}}<extra></extra>"
            ),
        ))
    if not relative and len(dates) <= _MAX_TOTAL_LABELS:
        fig.add_trace(go.Scatter(
            x=dates, y=totals, mode="text", text=[_amount(t, unit) for t in totals],
            textposition="top center", showlegend=False, hoverinfo="skip",
        ))
    fig.update_layout(
        template=_TEMPLATE, barmode="stack", bargap=0.45, height=460,
        margin=dict(l=20, r=20, t=30, b=20), hovermode="x unified",
        legend=bottom_legend(),
    )
    if all_runs:
        # A column per run, ticked with the night it was measured; the hover
        # names its release too.
        fig.update_xaxes(
            type="category", tickmode="array", tickvals=dates,
            ticktext=[label.split(" · ")[0] for label in dates], tickangle=-30,
            title_text="Night measured",
        )
    else:
        # The date axis the other trend views draw: one tick per nightly tag, and
        # a night without a run is a gap.
        fig.update_xaxes(
            type="date", tickmode="array", tickvals=dates,
            ticktext=[pd.Timestamp(d).strftime("%Y-%m-%d") for d in dates], tickangle=-30,
            title_text="Key4hep Nightly Tag",
        )
    top_y = 100 if relative else float(totals.max()) * 1.12
    fig.update_yaxes(
        title_text=f"Share of the {whole} (%)" if relative else y_title,
        range=[0, top_y or 1],
    )
    return fig

plot_region_timing

plot_region_timing(source: dict[str, dict] | str | Path | list[str | Path], *, labels: list[str] | None = None, show: str = 'both', attribution: str = 'at_location', top_n: int = 8, figsize: tuple[float, float] | None = None, exclude_events: list[int] | None = None, palette: list[str] | None = None, alpha: float = 0.85) -> Figure

Plot per-detector timing breakdown and/or per-event sequence for one or more runs.

Single run: a donut chart + sorted horizontal bar chart (breakdown panel) and/or a stacked-area sequence chart. Multiple runs: a grouped horizontal bar chart (breakdown) and/or a vertical stack of per-run stacked-area sequence charts.

Parameters:

Name	Type	Description	Default
source	dict[str, dict], str/Path, or list of str/Path	Pre-loaded dict from :func:~k4bench.analysis.loader.load_region_timing, a single log-dir path, or a list of log-dir paths.	required
labels	list[str] or None	Restrict to these run labels.	None
show	('both', 'breakdown', 'sequence')	Which panels to display.	"both"
attribution	('at_location', 'by_birth')	Which attribution to use.	"at_location"
top_n	int	Show the top n detectors by mean time; remaining are grouped into "Other".	8
figsize	(width, height) or None	Figure size in inches (converted to pixels at 96 dpi).	None
exclude_events	list[int] or None	Event numbers to exclude. Defaults to [0].	None

Returns:

Type	Description
Figure

Source code in k4bench/analysis/plots/region.py

def plot_region_timing(
    source: dict[str, dict] | str | Path | list[str | Path],
    *,
    labels: list[str] | None = None,
    show: str = "both",
    attribution: str = "at_location",
    top_n: int = 8,
    figsize: tuple[float, float] | None = None,
    exclude_events: list[int] | None = None,
    palette: list[str] | None = None,
    alpha: float = 0.85,
) -> go.Figure:
    """Plot per-detector timing breakdown and/or per-event sequence for one or more runs.

    Single run: a donut chart + sorted horizontal bar chart (breakdown panel) and/or
    a stacked-area sequence chart.  Multiple runs: a grouped horizontal bar chart
    (breakdown) and/or a vertical stack of per-run stacked-area sequence charts.

    Parameters
    ----------
    source : dict[str, dict], str/Path, or list of str/Path
        Pre-loaded dict from :func:`~k4bench.analysis.loader.load_region_timing`,
        a single log-dir path, or a list of log-dir paths.
    labels : list[str] or None
        Restrict to these run labels.
    show : {"both", "breakdown", "sequence"}
        Which panels to display.
    attribution : {"at_location", "by_birth"}
        Which attribution to use.
    top_n : int
        Show the top *n* detectors by mean time; remaining are grouped into ``"Other"``.
    figsize : (width, height) or None
        Figure size in inches (converted to pixels at 96 dpi).
    exclude_events : list[int] or None
        Event numbers to exclude.  Defaults to ``[0]``.

    Returns
    -------
    plotly.graph_objects.Figure
    """
    if show not in ("both", "breakdown", "sequence"):
        raise ValueError(f"show must be 'both', 'breakdown', or 'sequence', got {show!r}")
    if attribution not in ("at_location", "by_birth"):
        raise ValueError(f"attribution must be 'at_location' or 'by_birth', got {attribution!r}")

    if exclude_events is None:
        exclude_events = list(_DEFAULT_EXCLUDE_EVENTS)

    region_data = _ensure_region_data(source, labels=labels)
    if not region_data:
        raise ValueError(f"No region data found for labels={labels}.")

    label_list = list(region_data.keys())
    n = len(label_list)
    _pal = palette if palette is not None else _PALETTE

    # ------------------------------------------------------------------
    # Filter events and align DataFrames
    # ------------------------------------------------------------------
    filtered: dict[str, dict] = {}
    for lbl, data in region_data.items():
        ev_df = data["events"]
        mask  = ~ev_df["event_number"].isin(exclude_events)
        ev_filt = ev_df[mask].copy().reset_index(drop=True)
        if ev_filt.empty:
            raise ValueError(f"No events left after applying exclude_events for '{lbl}'.")
        time_df   = data[attribution]
        time_filt = (
            time_df.loc[time_df.index.isin(ev_filt["event_number"])]
            .reindex(ev_filt["event_number"].values)
        )
        missing_ev = time_filt.index[time_filt.isnull().any(axis=1)].tolist()
        if missing_ev:
            warnings.warn(
                f"'{lbl}': {len(missing_ev)} event(s) missing from time_df "
                f"(e.g. {missing_ev[:3]}); filling with 0.0 — means will be biased downward.",
                stacklevel=2,
            )
        time_filt = time_filt.fillna(0.0)
        filtered[lbl] = {"events": ev_filt, "time": time_filt}

    # ------------------------------------------------------------------
    # Top-N detectors — ranked by max-of-means across all runs so that
    # detectors dominant in any run are not folded into "Other".
    # ------------------------------------------------------------------
    means_per_run = [filtered[lbl]["time"].mean() for lbl in label_list]
    all_cols = sorted({col for s in means_per_run for col in s.index})
    union_max = pd.Series(
        {col: max(float(s.get(col, 0.0)) for s in means_per_run) for col in all_cols}
    )
    top_dets, all_dets_sorted = _region_top_n(union_max.to_frame().T, top_n)

    needs_other = len(all_dets_sorted) > top_n
    det_display = top_dets + (["Other"] if needs_other else [])
    det_colors: dict[str, str] = {
        det: _pal[i % len(_pal)] for i, det in enumerate(top_dets)
    }
    if "Other" in det_display:
        det_colors["Other"] = _OTHER_COLOR
    det_colors["Unaccounted"] = _UNACCOUNTED_COLOR

    show_breakdown = show in ("both", "breakdown")
    show_seq       = show in ("both", "sequence")

    # ------------------------------------------------------------------
    # Build figure layout
    # ------------------------------------------------------------------
    px_w = int(figsize[0] * 96) if figsize else 1200

    if n == 1:
        if show == "both":
            # Named constants so annotation positions can be derived analytically.
            # The donut is a circle, so its diameter is capped by the smaller of
            # its cell's width/height. Giving row 1 a larger *share* of the height
            # (and trimming the horizontal gap) lets the pie grow into the
            # otherwise-empty space without making the overall figure taller —
            # keep the total px_h fixed so the chart still fits without scrolling.
            _v_sp = 0.16
            _h_sp = _label_gap([*det_display, "Unaccounted"], px_w, 0.12)
            _r1_h, _r2_h = 0.50, 0.50
            fig = make_subplots(
                rows=2, cols=2,
                specs=[
                    [{"type": "domain"}, {"type": "xy"}],
                    [{"type": "xy", "colspan": 2}, None],
                ],
                row_heights=[_r1_h, _r2_h],
                vertical_spacing=_v_sp,
                horizontal_spacing=_h_sp,
            )
            px_h = figsize[1] * 96 if figsize else 1000
        elif show == "breakdown":
            fig = make_subplots(
                rows=1, cols=2,
                specs=[[{"type": "domain"}, {"type": "xy"}]],
                horizontal_spacing=0.08,
            )
            px_h = figsize[1] * 96 if figsize else 450
        else:
            fig = make_subplots(rows=1, cols=1)
            px_h = figsize[1] * 96 if figsize else 450
    else:
        if show == "both":
            seq_h_frac = 0.25
            bar_h_frac = max(0.15, 1.0 - seq_h_frac * n)
            row_heights = [bar_h_frac] + [seq_h_frac] * n
            n_rows_both = 1 + n
            v_spacing_both = min(0.06, 1.0 / (n_rows_both - 1)) if n_rows_both > 1 else 0.06
            fig = make_subplots(
                rows=n_rows_both, cols=1,
                row_heights=row_heights,
                vertical_spacing=v_spacing_both,
            )
            px_h = figsize[1] * 96 if figsize else (500 + 350 * n)
        elif show == "breakdown":
            fig = make_subplots(rows=1, cols=1)
            px_h = figsize[1] * 96 if figsize else 500
        else:
            v_spacing_seq = min(0.06, 1.0 / (n - 1)) if n > 1 else 0.06
            fig = make_subplots(
                rows=n, cols=1,
                shared_xaxes=True,
                vertical_spacing=v_spacing_seq,
            )
            px_h = figsize[1] * 96 if figsize else 350 * n

    # ------------------------------------------------------------------
    # Breakdown panel
    # ------------------------------------------------------------------
    if show_breakdown:
        lbl0     = label_list[0]
        time_df0 = filtered[lbl0]["time"]
        ev_df0   = filtered[lbl0]["events"]
        stacked0 = _build_stacked_arrays(time_df0, top_dets, all_dets_sorted)
        means0   = {det: float(arr.mean()) for det, arr in stacked0.items()}
        sems0    = {
            det: (float(arr.std(ddof=1)) / np.sqrt(len(arr)) if len(arr) > 1 else 0.0)
            for det, arr in stacked0.items()
        }
        mean_unacc = float(ev_df0["event_unaccounted_s"].mean())
        unacc_arr  = ev_df0["event_unaccounted_s"].to_numpy()
        sem_unacc  = float(unacc_arr.std(ddof=1)) / np.sqrt(len(unacc_arr)) if len(unacc_arr) > 1 else 0.0
        total_wall0 = float(ev_df0["event_wall_s"].mean())

        if n == 1 and show != "sequence":
            # Donut chart
            donut_cats = det_display + ["Unaccounted"]
            donut_vals = [means0.get(d, 0.0) for d in det_display] + [max(0.0, mean_unacc)]
            donut_clrs = [det_colors[c] for c in donut_cats]

            valid_idx = [i for i, v in enumerate(donut_vals) if v > 0]
            vv = [donut_vals[i] for i in valid_idx]
            vc = [donut_clrs[i] for i in valid_idx]
            vl = [donut_cats[i] for i in valid_idx]

            if vv:
                fig.add_trace(
                    go.Pie(
                        labels=vl, values=vv, hole=0.55,
                        marker=dict(colors=vc, line=dict(color="white", width=1.5)),
                        textinfo="percent",
                        textfont=dict(size=10),
                        hovertemplate="<b>%{label}</b><br>%{value:.3g} s<br>%{percent}<extra></extra>",
                        showlegend=False,
                    ),
                    row=1, col=1,
                )
                if show == "both":
                    # Centre of col-1 / row-1 in paper coordinates, derived from
                    # the make_subplots constants defined above.
                    # Col-1 occupies x ∈ [0, (1-_h_sp)/2]; centre = (1-_h_sp)/4
                    # Row-1 occupies y ∈ [_r2_h*(1-_v_sp)+_v_sp, 1.0]; centre = mid-point
                    _ann_x = (1 - _h_sp) / 4
                    _r1_bottom = _r2_h * (1 - _v_sp) + _v_sp
                    _ann_y = (_r1_bottom + 1.0) / 2
                else:
                    _ann_x, _ann_y = 0.23, 0.5
                fig.add_annotation(
                    text=f"μ = {total_wall0:.3g} s<br>per event",
                    x=_ann_x, y=_ann_y,
                    xref="paper", yref="paper",
                    xanchor="center", yanchor="middle",
                    showarrow=False,
                    font=dict(size=14, color="#333333"),
                    align="center",
                )

            # Horizontal bar (single run)
            bar_cats  = det_display + ["Unaccounted"]
            bar_means = [means0.get(d, 0.0) for d in det_display] + [max(0.0, mean_unacc)]
            bar_sems  = [sems0.get(d, 0.0)  for d in det_display] + [sem_unacc]
            bar_clrs  = [det_colors.get(d, _UNACCOUNTED_COLOR) for d in bar_cats]

            n_det_cats = len(det_display)
            order = sorted(range(n_det_cats), key=lambda i: bar_means[i], reverse=True)
            order.append(n_det_cats)
            s_cats  = [bar_cats[i]  for i in order]
            s_means = [bar_means[i] for i in order]
            s_sems  = [bar_sems[i]  for i in order]
            s_clrs  = [bar_clrs[i]  for i in order]

            fig.add_trace(
                go.Bar(
                    y=s_cats, x=s_means,
                    orientation="h",
                    marker_color=s_clrs,
                    marker_line_width=0,
                    error_x=dict(type="data", array=s_sems, visible=True,
                                 thickness=0.8, color="#555555"),
                    showlegend=False,
                    hovertemplate="%{y}<br><b>%{x:.3g} s</b> ± %{error_x.array:.2g} s<extra></extra>",
                ),
                row=1, col=2,
            )
            x_right = max(v + s for v, s in zip(s_means, s_sems)) if s_means else 1.0
            fig.add_trace(
                go.Scatter(
                    x=[v + s + x_right * 0.03 for v, s in zip(s_means, s_sems)],
                    y=s_cats,
                    mode="text",
                    text=[f"{v:.3g} s" for v in s_means],
                    textposition="middle right",
                    textfont=dict(size=10, color="#333333"),
                    showlegend=False,
                    hoverinfo="skip",
                ),
                row=1, col=2,
            )
            fig.update_xaxes(title_text="Mean time per event (s)",
                             range=[0, x_right * 1.45], row=1, col=2)
            fig.update_yaxes(autorange="reversed", row=1, col=2)

        else:
            # Grouped bar (multi-run)
            all_run_means: dict[str, dict[str, float]] = {}
            all_run_unacc: dict[str, float] = {}
            for lbl in label_list:
                td = filtered[lbl]["time"]
                ed = filtered[lbl]["events"]
                st = _build_stacked_arrays(td, top_dets, all_dets_sorted)
                all_run_means[lbl] = {det: float(arr.mean()) for det, arr in st.items()}
                all_run_unacc[lbl] = max(0.0, float(ed["event_unaccounted_s"].mean()))

            all_bar_dets = det_display + ["Unaccounted"]
            for run_i, lbl in enumerate(label_list):
                run_vals = [all_run_means[lbl].get(d, 0.0) for d in det_display]
                run_vals.append(all_run_unacc[lbl])
                fig.add_trace(
                    go.Bar(
                        y=all_bar_dets, x=run_vals,
                        orientation="h",
                        name=lbl,
                        marker_color=_pal[run_i % len(_pal)],
                        opacity=alpha,
                        marker_line_width=0,
                        hovertemplate=f"<b>{lbl}</b><br>%{{y}}: %{{x:.3g}} s<extra></extra>",
                    ),
                    row=1, col=1,
                )
            fig.update_xaxes(title_text="Mean time per event (s)", row=1, col=1)
            fig.update_yaxes(autorange="reversed", row=1, col=1)
            fig.update_layout(barmode="group")

    # ------------------------------------------------------------------
    # Sequence panel(s)
    # ------------------------------------------------------------------
    if show_seq:
        for run_i, lbl in enumerate(label_list):
            time_df = filtered[lbl]["time"]
            ev_df   = filtered[lbl]["events"]

            event_nums = time_df.index.to_numpy()
            ev_idx     = ev_df.set_index("event_number").reindex(event_nums)
            wall_times = ev_idx["event_wall_s"].to_numpy()
            _unacc_raw = ev_idx["event_unaccounted_s"].to_numpy()
            _neg_frac  = (_unacc_raw < -1e-6).mean()
            if _neg_frac > 0.05:
                warnings.warn(
                    f"'{lbl}': {_neg_frac:.0%} of events have negative unaccounted time "
                    f"(min={_unacc_raw.min():.3g} s). Region timing sum may exceed wall time.",
                    stacklevel=2,
                )
            unaccounted = np.maximum(0.0, _unacc_raw)

            stacked     = _build_stacked_arrays(time_df, top_dets, all_dets_sorted)
            stack_order = sorted(stacked, key=lambda d: stacked[d].mean(), reverse=True)

            if n == 1:
                seq_rc = (2, 1) if show == "both" else (1, 1)
            else:
                seq_row = (run_i + 2) if show == "both" else (run_i + 1)
                seq_rc  = (seq_row, 1)

            for det in stack_order:
                color = det_colors.get(det, _OTHER_COLOR)
                fig.add_trace(
                    go.Scatter(
                        x=event_nums, y=stacked[det],
                        name=det,
                        mode="lines",
                        line=dict(width=0, color=color),
                        fillcolor=_hex_to_rgba(color, alpha),
                        fill="tonexty",
                        stackgroup=f"stack_{lbl}",
                        legendgroup=det,
                        showlegend=(run_i == 0),
                        hovertemplate=f"<b>{det}</b><br>event: %{{x}}<br>%{{y:.4g}} s<extra></extra>",
                    ),
                    row=seq_rc[0], col=seq_rc[1],
                )

            fig.add_trace(
                go.Scatter(
                    x=event_nums, y=unaccounted,
                    name="Unaccounted",
                    mode="lines",
                    line=dict(width=0, color=_UNACCOUNTED_COLOR),
                    fillcolor=_hex_to_rgba(_UNACCOUNTED_COLOR, alpha),
                    fill="tonexty",
                    stackgroup=f"stack_{lbl}",
                    legendgroup="Unaccounted",
                    showlegend=(run_i == 0),
                    hovertemplate="<b>Unaccounted</b><br>event: %{x}<br>%{y:.4g} s<extra></extra>",
                ),
                row=seq_rc[0], col=seq_rc[1],
            )

            fig.add_trace(
                go.Scatter(
                    x=event_nums, y=wall_times,
                    name="Wall time",
                    mode="lines",
                    line=dict(color="black", width=0.9, dash="dash"),
                    opacity=0.6,
                    legendgroup="Wall time",
                    showlegend=(run_i == 0),
                    hovertemplate="<b>Wall time</b><br>event: %{x}<br>%{y:.4g} s<extra></extra>",
                ),
                row=seq_rc[0], col=seq_rc[1],
            )

            fig.update_xaxes(title_text="Event number", row=seq_rc[0], col=seq_rc[1])
            if n > 1:
                fig.update_yaxes(title_text=lbl, title_font=dict(size=9),
                                 row=seq_rc[0], col=seq_rc[1])
            else:
                fig.update_yaxes(title_text="Time (s)", row=seq_rc[0], col=seq_rc[1])

    # ------------------------------------------------------------------
    # Layout
    fig.update_layout(
        template=_TEMPLATE,
        width=px_w,
        height=int(px_h),
        legend=dict(
            orientation="v",
            yanchor="top", y=1.0,
            xanchor="left", x=1.01,
            font=dict(size=9),
        ),
        margin=dict(l=20, r=160, t=30, b=40),
    )

    return fig