k4bench.analysis.loader

k4bench.analysis.loader

Load benchmark results and per-event timing data for analysis.

with_cpu_efficiency

with_cpu_efficiency(df: DataFrame) -> DataFrame

Return df with total-CPU/wall cpu_efficiency when measurable.

CPU time means user plus system time everywhere. If any component is absent, the efficiency is unknown and no derived column is attached.

Source code in k4bench/analysis/loader.py

def with_cpu_efficiency(df: pd.DataFrame) -> pd.DataFrame:
    """Return *df* with total-CPU/wall ``cpu_efficiency`` when measurable.

    CPU time means user plus system time everywhere. If any component is
    absent, the efficiency is unknown and no derived column is attached.
    """
    if not {"user_cpu_s", "sys_cpu_s", "wall_time_s"} <= set(df.columns):
        return df
    out = df.copy()
    total_cpu = (
        pd.to_numeric(out["user_cpu_s"], errors="coerce")
        + pd.to_numeric(out["sys_cpu_s"], errors="coerce")
    )
    wall = pd.to_numeric(out["wall_time_s"], errors="coerce")
    out["cpu_efficiency"] = total_cpu / wall.replace(0, float("nan"))
    return out

failed_config_mask

failed_config_mask(df: DataFrame) -> Series

Return a boolean mask for configs that did not exit cleanly.

A non-zero, missing, or non-numeric returncode is a failure. Frames from older result schemas that have no returncode column remain usable: their success state is unknown, not known-bad, so the returned mask is all False. Source-column provenance preserves that distinction when old and new result frames are later concatenated. The mask always preserves df's index.

Source code in k4bench/analysis/loader.py

def failed_config_mask(df: pd.DataFrame) -> pd.Series:
    """Return a boolean mask for configs that did not exit cleanly.

    A non-zero, missing, or non-numeric ``returncode`` is a failure.  Frames
    from older result schemas that have no ``returncode`` column remain usable:
    their success state is unknown, not known-bad, so the returned mask is all
    ``False``. Source-column provenance preserves that distinction when old and
    new result frames are later concatenated. The mask always preserves *df*'s
    index.
    """
    if "returncode" not in df.columns:
        return pd.Series(False, index=df.index, dtype=bool)

    returncodes = pd.to_numeric(df["returncode"], errors="coerce")
    recorded = (
        df[_RETURNCODE_RECORDED].astype("boolean").fillna(True)
        if _RETURNCODE_RECORDED in df.columns
        else pd.Series(True, index=df.index, dtype=bool)
    )
    # Compare before filling so this also works for nullable unsigned and
    # boolean dtypes, which cannot represent a ``-1`` fill sentinel.
    return (recorded & returncodes.ne(0).fillna(True)).astype(bool)

config_keys

config_keys(df: DataFrame | None) -> set[tuple[str, str]]

Return every (run_id, label) pair represented by df.

Source code in k4bench/analysis/loader.py

def config_keys(df: pd.DataFrame | None) -> set[tuple[str, str]]:
    """Return every ``(run_id, label)`` pair represented by *df*."""
    if (
        df is None
        or df.empty
        or not {"run_id", "label"} <= set(df.columns)
    ):
        return set()
    return {
        (str(run_id), str(label))
        for run_id, label in df[["run_id", "label"]].itertuples(
            index=False, name=None,
        )
    }

failed_config_keys

failed_config_keys(results_df: DataFrame | None) -> set[tuple[str, str]]

Return failed (run_id, label) pairs from a result-history frame.

Source code in k4bench/analysis/loader.py

def failed_config_keys(
    results_df: pd.DataFrame | None,
) -> set[tuple[str, str]]:
    """Return failed ``(run_id, label)`` pairs from a result-history frame."""
    if (
        results_df is None
        or results_df.empty
        or not {"run_id", "label"} <= set(results_df.columns)
    ):
        return set()
    failed = results_df.loc[failed_config_mask(results_df), ["run_id", "label"]]
    return {
        (str(run_id), str(label))
        for run_id, label in failed.itertuples(index=False, name=None)
    }

judgeable_config_keys

judgeable_config_keys(results_df: DataFrame | None) -> set[tuple[str, str]]

Result config-nights with a recorded success or legacy-unknown status.

Source code in k4bench/analysis/loader.py

def judgeable_config_keys(
    results_df: pd.DataFrame | None,
) -> set[tuple[str, str]]:
    """Result config-nights with a recorded success or legacy-unknown status."""
    return config_keys(results_df) - failed_config_keys(results_df)

config_rows_for_keys

config_rows_for_keys(df: DataFrame | None, keys: set[tuple[str, str]]) -> DataFrame | None

Keep rows whose normalized (run_id, label) pair is in keys.

The helper is the shared join primitive for result-backed derived data. Keeping it separate from the policy helpers below lets callers retain recorded failed rows for diagnostic plots while still rejecting orphaned event/region files.

Source code in k4bench/analysis/loader.py

def config_rows_for_keys(
    df: pd.DataFrame | None,
    keys: set[tuple[str, str]],
) -> pd.DataFrame | None:
    """Keep rows whose normalized ``(run_id, label)`` pair is in *keys*.

    The helper is the shared join primitive for result-backed derived data.
    Keeping it separate from the policy helpers below lets callers retain
    recorded failed rows for diagnostic plots while still rejecting orphaned
    event/region files.
    """
    if (
        df is None
        or df.empty
        or not {"run_id", "label"} <= set(df.columns)
    ):
        return df
    row_keys = pd.MultiIndex.from_arrays([
        df["run_id"].astype(str),
        df["label"].astype(str),
    ])
    keep = (
        row_keys.isin(pd.MultiIndex.from_tuples(sorted(keys)))
        if keys else [False] * len(df)
    )
    return df.loc[keep].copy()

recorded_config_rows

recorded_config_rows(df: DataFrame | None, results_df: DataFrame | None) -> DataFrame | None

Keep metric rows backed by any result row, including failed configs.

Source code in k4bench/analysis/loader.py

def recorded_config_rows(
    df: pd.DataFrame | None,
    results_df: pd.DataFrame | None,
) -> pd.DataFrame | None:
    """Keep metric rows backed by any result row, including failed configs."""
    return config_rows_for_keys(df, config_keys(results_df))

judgeable_config_rows

judgeable_config_rows(df: DataFrame | None, results_df: DataFrame | None) -> DataFrame | None

Keep metric rows backed by a non-failed result config-night.

Derived event and region frames do not carry returncode themselves, so they inherit validity from the result frame through (run_id, label). This also rejects orphaned partial files whose config never wrote a result row, while retaining healthy sibling configs from the same run.

Source code in k4bench/analysis/loader.py

def judgeable_config_rows(
    df: pd.DataFrame | None,
    results_df: pd.DataFrame | None,
) -> pd.DataFrame | None:
    """Keep metric rows backed by a non-failed result config-night.

    Derived event and region frames do not carry ``returncode`` themselves, so
    they inherit validity from the result frame through ``(run_id, label)``.
    This also rejects orphaned partial files whose config never wrote a result
    row, while retaining healthy sibling configs from the same run.
    """
    return config_rows_for_keys(df, judgeable_config_keys(results_df))

judgeable_config_data

judgeable_config_data(data: Mapping[str, _T] | None, results_df: DataFrame | None) -> dict[str, _T] | None

Filter a current-run label -> payload mapping to successful configs.

Current event and region files are dict-backed rather than tabular. A label is kept only when the current result frame contains at least one non-failed row for it; failed configs and orphaned partial files therefore follow the same policy as their historical DataFrame counterparts.

Source code in k4bench/analysis/loader.py

def judgeable_config_data(
    data: Mapping[str, _T] | None,
    results_df: pd.DataFrame | None,
) -> dict[str, _T] | None:
    """Filter a current-run ``label -> payload`` mapping to successful configs.

    Current event and region files are dict-backed rather than tabular.  A
    label is kept only when the current result frame contains at least one
    non-failed row for it; failed configs and orphaned partial files therefore
    follow the same policy as their historical DataFrame counterparts.
    """
    if data is None:
        return None
    if results_df is None or results_df.empty or "label" not in results_df.columns:
        return {}
    all_labels = {str(label) for label in results_df["label"].dropna()}
    failed_labels = {
        str(label)
        for label in results_df.loc[failed_config_mask(results_df), "label"].dropna()
    }
    labels = all_labels - failed_labels
    return {label: value for label, value in data.items() if str(label) in labels}

load_results

load_results(log_dir: str | Path, labels: list[str] | None = None) -> DataFrame

Load benchmark results from a log directory into a DataFrame.

Each {label}_results.csv file written by k4bench is loaded and concatenated into a single DataFrame.

Parameters:

Name	Type	Description	Default
log_dir	str or Path	Directory containing *_results.csv files.	required
labels	list[str] or None	Load only these run labels. Loads all *_results.csv files when None.	None

Returns:

Type	Description
DataFrame	One row per run. Float columns are cast to float64; integer columns that may contain NaN use nullable Int64.

Source code in k4bench/analysis/loader.py

def load_results(log_dir: str | Path, labels: list[str] | None = None) -> pd.DataFrame:
    """Load benchmark results from a log directory into a DataFrame.

    Each ``{label}_results.csv`` file written by ``k4bench`` is loaded and
    concatenated into a single DataFrame.

    Parameters
    ----------
    log_dir : str or Path
        Directory containing ``*_results.csv`` files.
    labels : list[str] or None
        Load only these run labels.  Loads all ``*_results.csv`` files when ``None``.

    Returns
    -------
    pd.DataFrame
        One row per run. Float columns are cast to ``float64``;
        integer columns that may contain NaN use nullable ``Int64``.
    """
    log_dir = Path(log_dir)
    paths = [path for path, _ in _label_files(log_dir, RESULTS_SUFFIX, labels, "result")]

    if not paths:
        raise ValueError(f"No *{RESULTS_SUFFIX} files found in '{log_dir}'.")

    frames = [pd.read_csv(p) for p in paths]
    if not all("returncode" in frame.columns for frame in frames):
        # Preserve whether each source file actually recorded this field before
        # concat turns an absent old-schema column into an ordinary missing
        # value.  The marker is omitted from uniform modern loads and is
        # otherwise harmless internal metadata carried through trend frames.
        for frame in frames:
            frame[_RETURNCODE_RECORDED] = "returncode" in frame.columns
    df = pd.concat(frames, ignore_index=True)

    float_cols = [
        "wall_time_s",
        "user_cpu_s",
        "sys_cpu_s",
        "peak_rss_mb",
        "peak_vmem_mb",
        "output_size_mb",
        "events_per_sec",
    ]
    int_cols = [
        "returncode", "n_events",
        "major_page_faults", "voluntary_ctx_switches", "involuntary_ctx_switches",
    ]

    for col in float_cols:
        if col in df.columns:
            df[col] = pd.to_numeric(df[col], errors="coerce")

    for col in int_cols:
        if col in df.columns:
            df[col] = pd.to_numeric(df[col], errors="coerce").astype("Int64")

    return df

load_event_timing

load_event_timing(log_dir: str | Path, labels: list[str] | None = None) -> dict[str, DataFrame]

Load per-event timing JSON files from a log directory.

Each {label}_events.json file written by the k4BenchTimingAction plugin is parsed into a DataFrame.

Parameters:

Name	Type	Description	Default
log_dir	str or Path	Directory containing *_events.json files.	required
labels	list[str] or None	Load only these run labels. If None, all *_events.json files in log_dir are loaded.	None

Returns:

Type	Description
dict[str, DataFrame]	Maps label → DataFrame with columns event_number, event_time_s, rss_begin_mb, rss_end_mb, rss_delta_mb. Newer files also supply rss_anon_begin_mb, rss_anon_end_mb, rss_file_end_mb; each optional column is added only when its key is present.

Raises:

Type	Description
ValueError	For a JSON root that is not an object, a malformed or unsupported schema_version (:func:~k4bench.plugin.schema.validate_event_schema), missing required keys, or mismatched array lengths. A file without schema_version is the legacy unversioned format and loads as usual.

Source code in k4bench/analysis/loader.py

def load_event_timing(
    log_dir: str | Path,
    labels: list[str] | None = None,
) -> dict[str, pd.DataFrame]:
    """Load per-event timing JSON files from a log directory.

    Each ``{label}_events.json`` file written by the k4BenchTimingAction
    plugin is parsed into a DataFrame.

    Parameters
    ----------
    log_dir : str or Path
        Directory containing ``*_events.json`` files.
    labels : list[str] or None
        Load only these run labels. If None, all ``*_events.json`` files
        in *log_dir* are loaded.

    Returns
    -------
    dict[str, pd.DataFrame]
        Maps label → DataFrame with columns
        ``event_number``, ``event_time_s``, ``rss_begin_mb``,
        ``rss_end_mb``, ``rss_delta_mb``. Newer files also supply
        ``rss_anon_begin_mb``, ``rss_anon_end_mb``, ``rss_file_end_mb``;
        each optional column is added only when its key is present.

    Raises
    ------
    ValueError
        For a JSON root that is not an object, a malformed or unsupported
        ``schema_version``
        (:func:`~k4bench.plugin.schema.validate_event_schema`), missing
        required keys, or mismatched array lengths. A file without
        ``schema_version`` is the legacy unversioned format and loads as usual.
    """
    out: dict[str, pd.DataFrame] = {}
    for path, label in _label_files(Path(log_dir), EVENTS_SUFFIX, labels, "event"):
        with path.open() as f:
            raw = json.load(f)
        validate_event_schema(raw, source=path)
        _required = ["event_numbers", "event_times_s", "event_rss_begin_mb", "event_rss_end_mb"]
        _missing = [k for k in _required if k not in raw]
        if _missing:
            raise ValueError(f"{path} missing keys: {_missing}")
        optional = {
            "event_rss_anon_begin_mb": "rss_anon_begin_mb",
            "event_rss_anon_end_mb": "rss_anon_end_mb",
            "event_rss_file_end_mb": "rss_file_end_mb",
        }
        lengths = {k: len(raw[k]) for k in [*_required, *optional] if k in raw}
        if len(set(lengths.values())) > 1:
            raise ValueError(f"{path} has mismatched array lengths: {lengths}")
        df = pd.DataFrame(
            {
                "event_number": raw["event_numbers"],
                "event_time_s": raw["event_times_s"],
                "rss_begin_mb": raw["event_rss_begin_mb"],
                "rss_end_mb":   raw["event_rss_end_mb"],
            }
        )
        df["rss_delta_mb"] = df["rss_end_mb"] - df["rss_begin_mb"]
        for key, column in optional.items():
            if key in raw:
                df[column] = raw[key]
        out[label] = df

    return out

load_region_timing

load_region_timing(log_dir: str | Path, labels: list[str] | None = None) -> dict[str, dict]

Load per-region timing JSON files from a log directory.

Each {label}_regions.json file written by the k4BenchRegionTimingAction plugin is parsed into structured data.

Parameters:

Name	Type	Description	Default
log_dir	str or Path	Directory containing *_regions.json files.	required
labels	list[str] or None	Load only these run labels. If None, all *_regions.json files in log_dir are loaded.	None

Returns:

Type

Description

dict[str, dict]

Maps label → dict with keys: "meta": dict with schema_version, timer, overhead_ns, detectors, lv_counts. "events": DataFrame with columns event_number, event_wall_s, event_region_sum_s, event_unaccounted_s. "at_location": DataFrame indexed by event_number, one column per top-level detector (seconds), time charged to where the Geant4 step physically occurred. "by_birth": same shape as at_location, time charged to the detector where the primary track was created.

Source code in k4bench/analysis/loader.py

def load_region_timing(
    log_dir: str | Path,
    labels: list[str] | None = None,
) -> dict[str, dict]:
    """Load per-region timing JSON files from a log directory.

    Each ``{label}_regions.json`` file written by the k4BenchRegionTimingAction
    plugin is parsed into structured data.

    Parameters
    ----------
    log_dir : str or Path
        Directory containing ``*_regions.json`` files.
    labels : list[str] or None
        Load only these run labels.  If ``None``, all ``*_regions.json`` files
        in *log_dir* are loaded.

    Returns
    -------
    dict[str, dict]
        Maps label → dict with keys:

        - ``"meta"``: dict with schema_version, timer, overhead_ns, detectors,
          lv_counts.
        - ``"events"``: DataFrame with columns ``event_number``,
          ``event_wall_s``, ``event_region_sum_s``, ``event_unaccounted_s``.
        - ``"at_location"``: DataFrame indexed by ``event_number``, one column
          per top-level detector (seconds), time charged to where the Geant4
          step physically occurred.
        - ``"by_birth"``: same shape as ``at_location``, time charged to the
          detector where the primary track was created.
    """
    out: dict[str, dict] = {}
    for path, label in _label_files(Path(log_dir), REGIONS_SUFFIX, labels, "region"):
        with path.open() as f:
            raw = json.load(f)
        validate_region_schema(raw, source=path)

        _required = [
            "event_numbers", "event_wall_seconds",
            "event_region_sum_seconds", "event_unaccounted_seconds",
            "at_location_seconds", "by_birth_seconds",
        ]
        _missing = [k for k in _required if k not in raw]
        if _missing:
            raise ValueError(f"{path} missing keys: {_missing}")

        n_ev = len(raw["event_numbers"])
        if len(set(raw["event_numbers"])) != n_ev:
            raise ValueError(f"{path}: event_numbers contains duplicates")
        for k in ["event_wall_seconds", "event_region_sum_seconds",
                  "event_unaccounted_seconds", "at_location_seconds", "by_birth_seconds"]:
            if len(raw[k]) != n_ev:
                raise ValueError(f"{path}: array length mismatch for '{k}'")

        events_df = pd.DataFrame({
            "event_number":       raw["event_numbers"],
            "event_wall_s":       raw["event_wall_seconds"],
            "event_region_sum_s": raw["event_region_sum_seconds"],
            "event_unaccounted_s": raw["event_unaccounted_seconds"],
        })

        ev_index = pd.Index(raw["event_numbers"], name="event_number")
        at_loc_df   = pd.DataFrame(raw["at_location_seconds"], index=ev_index).fillna(0.0)
        by_birth_df = pd.DataFrame(raw["by_birth_seconds"],    index=ev_index).fillna(0.0)

        declared = raw.get("indexed_top_level_detectors", [])
        if declared:
            extra_at  = [c for c in at_loc_df.columns  if c not in declared]
            extra_by  = [c for c in by_birth_df.columns if c not in declared]
            at_loc_df   = at_loc_df.reindex(  columns=declared + extra_at,  fill_value=0.0)
            by_birth_df = by_birth_df.reindex( columns=declared + extra_by, fill_value=0.0)

        # Step counts per detector per event (interval_counts field, optional).
        # Columns include all detector names plus "unattributed" for steps that
        # could not be assigned to a top-level detector element.
        raw_steps = raw.get("interval_counts", [])
        if raw_steps and len(raw_steps) == n_ev:
            steps_df: pd.DataFrame | None = (
                pd.DataFrame(raw_steps, index=ev_index).fillna(0).astype(float)
            )
        else:
            steps_df = None

        out[label] = {
            "meta": {
                "schema_version":    raw.get("schema_version", 1),
                "attribution_method": raw.get("attribution", "dd4hep_top_level_detelement"),
                "timer":             raw.get("timer", "unknown"),
                "overhead_ns":       raw.get("per_step_timer_overhead_ns"),
                "detectors":         raw.get("indexed_top_level_detectors", []),
                "lv_counts":         raw.get("indexed_top_level_detector_lv_counts", {}),
            },
            "events":      events_df,
            "at_location": at_loc_df,
            "by_birth":    by_birth_df,
            "steps":       steps_df,   # None when interval_counts absent in JSON
        }

    if not out:
        if labels is not None:
            raise ValueError(
                f"No region files found for labels={labels} in '{log_dir}'."
            )
        raise ValueError(f"No *{REGIONS_SUFFIX} files found in '{log_dir}'.")

    return out

load_component_timing

load_component_timing(log_dir: str | Path, labels: list[str] | None = None) -> dict[str, ComponentTiming]

Load the per-component files the k4BenchAuditor writes for k4run jobs.

Parameters:

Name	Type	Description	Default
log_dir	str or Path	Directory containing *_components.json files.	required
labels	list[str] or None	Load only these run labels. If None, every *_components.json file in log_dir is loaded.	None

Returns:

Type	Description
dict[str, ComponentTiming]	Maps label → its :class:~k4bench.analysis.components.ComponentTiming. Empty when log_dir holds no component files (a ddsim run).

Raises:

Type	Description
ValueError	For a requested label without a file, an unsupported schema_version (:func:~k4bench.plugin.schema.validate_component_schema), or a file whose tables do not match its component and event lists; the first such file stops the load. :func:load_each_component_timing reads each file on its own instead.

Source code in k4bench/analysis/loader.py

def load_component_timing(
    log_dir: str | Path,
    labels: list[str] | None = None,
) -> dict[str, ComponentTiming]:
    """Load the per-component files the k4BenchAuditor writes for k4run jobs.

    Parameters
    ----------
    log_dir : str or Path
        Directory containing ``*_components.json`` files.
    labels : list[str] or None
        Load only these run labels. If ``None``, every ``*_components.json``
        file in *log_dir* is loaded.

    Returns
    -------
    dict[str, ComponentTiming]
        Maps label → its :class:`~k4bench.analysis.components.ComponentTiming`.
        Empty when *log_dir* holds no component files (a ddsim run).

    Raises
    ------
    ValueError
        For a requested label without a file, an unsupported
        ``schema_version`` (:func:`~k4bench.plugin.schema.validate_component_schema`),
        or a file whose tables do not match its component and event lists;
        the first such file stops the load. :func:`load_each_component_timing`
        reads each file on its own instead.
    """
    out: dict[str, ComponentTiming] = {}
    for path, label in _label_files(Path(log_dir), COMPONENTS_SUFFIX, labels, "component"):
        with path.open() as f:
            raw = json.load(f)
        validate_component_schema(raw, source=path)
        out[label] = ComponentTiming.from_json(raw, source=path)
    return out

unreadable_reason

unreadable_reason(path: Path, exc: BaseException) -> str

Why path could not be read, from exc: the file's name, the error's class and its message, with the file's full path left out and the message cut short, so it shows none of the file's contents at length.

Source code in k4bench/analysis/loader.py

def unreadable_reason(path: Path, exc: BaseException) -> str:
    """Why *path* could not be read, from *exc*: the file's name, the error's
    class and its message, with the file's full path left out and the
    message cut short, so it shows none of the file's contents at length."""
    if isinstance(exc, OSError) and exc.strerror:
        message = exc.strerror
    else:
        message = str(exc).replace(f"{path}: ", "").replace(str(path), path.name)
    if len(message) > _MAX_REASON_CHARS:
        message = message[: _MAX_REASON_CHARS - 1] + "…"
    return f"{path.name}: {type(exc).__name__}: {message}"

load_each_component_timing

load_each_component_timing(log_dir: str | Path) -> tuple[dict[str, ComponentTiming], dict[str, str]]

Every label's _components.json in log_dir that could be read (:func:load_component_timing), and why each other could not. Both empty when log_dir holds no component files (a ddsim run).

Source code in k4bench/analysis/loader.py

def load_each_component_timing(
    log_dir: str | Path,
) -> tuple[dict[str, ComponentTiming], dict[str, str]]:
    """Every label's ``_components.json`` in *log_dir* that could be read
    (:func:`load_component_timing`), and why each other could not. Both empty
    when *log_dir* holds no component files (a ddsim run)."""
    return _load_each(load_component_timing, Path(log_dir), COMPONENTS_SUFFIX)

load_each_component_heap

load_each_component_heap(log_dir: str | Path) -> tuple[dict[str, ComponentHeap], dict[str, str]]

Every label's _heap.json in log_dir that could be read (:func:load_component_heap), and why each other could not. Both empty when log_dir holds no heap files (no memory pass ran).

Source code in k4bench/analysis/loader.py

def load_each_component_heap(
    log_dir: str | Path,
) -> tuple[dict[str, ComponentHeap], dict[str, str]]:
    """Every label's ``_heap.json`` in *log_dir* that could be read
    (:func:`load_component_heap`), and why each other could not. Both empty
    when *log_dir* holds no heap files (no memory pass ran)."""
    return _load_each(load_component_heap, Path(log_dir), HEAP_SUFFIX)

load_component_heap

load_component_heap(log_dir: str | Path, labels: list[str] | None = None) -> dict[str, ComponentHeap]

Load the heap files the k4run benchmark's memory pass writes.

Parameters:

Name	Type	Description	Default
log_dir	str or Path	Directory containing *_heap.json files.	required
labels	list[str] or None	Load only these run labels. If None, every *_heap.json file in log_dir is loaded.	None

Returns:

Type	Description
dict[str, ComponentHeap]	Maps label → its :class:~k4bench.analysis.heap.ComponentHeap. Empty when log_dir holds no heap files (no memory pass ran).

Raises:

Type	Description
ValueError	For a requested label without a file, an unsupported schema_version (:func:~k4bench.plugin.schema.validate_heap_schema), or a file whose tables do not match its component and event lists.

Source code in k4bench/analysis/loader.py

def load_component_heap(
    log_dir: str | Path,
    labels: list[str] | None = None,
) -> dict[str, ComponentHeap]:
    """Load the heap files the k4run benchmark's memory pass writes.

    Parameters
    ----------
    log_dir : str or Path
        Directory containing ``*_heap.json`` files.
    labels : list[str] or None
        Load only these run labels. If ``None``, every ``*_heap.json`` file in
        *log_dir* is loaded.

    Returns
    -------
    dict[str, ComponentHeap]
        Maps label → its :class:`~k4bench.analysis.heap.ComponentHeap`.
        Empty when *log_dir* holds no heap files (no memory pass ran).

    Raises
    ------
    ValueError
        For a requested label without a file, an unsupported
        ``schema_version`` (:func:`~k4bench.plugin.schema.validate_heap_schema`),
        or a file whose tables do not match its component and event lists.
    """
    out: dict[str, ComponentHeap] = {}
    for path, label in _label_files(Path(log_dir), HEAP_SUFFIX, labels, "heap"):
        with path.open() as f:
            raw = json.load(f)
        validate_heap_schema(raw, source=path)
        out[label] = ComponentHeap.from_json(raw, source=path)
    return out

run_suite

run_suite(run_dir: str | Path) -> str

The suite of the run in run_dir.

A nightly run records it in its run_info.json. A run made with the k4bench CLI, or a nightly run older than that field, records none: it is a reco run exactly when the k4run auditor left a _components.json.

Source code in k4bench/analysis/loader.py

def run_suite(run_dir: str | Path) -> str:
    """The suite of the run in *run_dir*.

    A nightly run records it in its ``run_info.json``. A run made with the
    ``k4bench`` CLI, or a nightly run older than that field, records none: it
    is a reco run exactly when the k4run auditor left a ``_components.json``.
    """
    try:
        run_info = json.loads((Path(run_dir) / RUN_INFO).read_text())
    except (OSError, ValueError):
        run_info = None
    suite = run_info.get("suite") if isinstance(run_info, dict) else None
    if suite in SUITES:
        return suite
    return RECO if labelled_files(Path(run_dir), COMPONENTS_SUFFIX) else SIM