Merge pull request #11 from SteSeg/report_generation

Copy `specifications` and `run_metadata` into `benchmark_results.h5`

Author: SteSeg

scripts/backfill_results_metadata.py

@@ -0,0 +1,115 @@
+#!/usr/bin/env python3
+"""Backfill specifications and run metadata into existing results files."""
+
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+
+import h5py
+import numpy as np
+import yaml
+
+
+def _write_spec_snapshot(handle: h5py.File, spec_path: Path, benchmark_name: str | None) -> None:
+    spec_yaml = spec_path.read_text(encoding="utf-8")
+    spec_bytes = spec_yaml.encode("utf-8")
+
+    if "specifications" in handle:
+        del handle["specifications"]
+    group = handle.create_group("specifications")
+    group.attrs["format"] = "yaml"
+    if benchmark_name:
+        group.attrs["benchmark_name"] = benchmark_name
+    group.create_dataset("yaml", data=np.bytes_(spec_bytes))
+
+
+def _write_run_metadata(
+    handle: h5py.File,
+    code_name: str | None,
+    code_version: str | None,
+    nuclear_data_name: str | None,
+    nuclear_data_version: str | None,
+    geometry: str | None,
+) -> None:
+    if "run_metadata" in handle:
+        del handle["run_metadata"]
+    group = handle.create_group("run_metadata")
+
+    if code_name:
+        group.attrs["code_name"] = code_name
+    if code_version:
+        group.attrs["code_version"] = code_version
+    if nuclear_data_name:
+        group.attrs["nuclear_data_name"] = nuclear_data_name
+    if nuclear_data_version:
+        group.attrs["nuclear_data_version"] = nuclear_data_version
+    if geometry:
+        group.attrs["geometry"] = geometry
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Backfill specifications and run metadata into an OFB results file."
+    )
+    parser.add_argument("--file", type=Path, required=True, help="Path to results .h5 file")
+    parser.add_argument("--spec-file", type=Path, required=True, help="Path to specifications.yaml")
+    parser.add_argument(
+        "--kind",
+        choices=["experiment", "calculation"],
+        required=True,
+        help="Result kind (controls which metadata are written)",
+    )
+    parser.add_argument("--benchmark-name", type=str, default=None, help="Benchmark name label")
+    parser.add_argument("--code-name", type=str, default=None, help="Code name (calculation only)")
+    parser.add_argument("--code-version", type=str, default=None, help="Code version (calculation only)")
+    parser.add_argument(
+        "--nuclear-data-name",
+        type=str,
+        default=None,
+        help="Nuclear data library name (calculation only)",
+    )
+    parser.add_argument(
+        "--nuclear-data-version",
+        type=str,
+        default=None,
+        help="Nuclear data library version (calculation only)",
+    )
+    parser.add_argument(
+        "--geometry",
+        type=str,
+        default=None,
+        choices=["cad", "csg"],
+        help="Geometry type (calculation only)",
+    )
+    args = parser.parse_args()
+
+    if not args.file.exists():
+        raise FileNotFoundError(f"Results file not found: {args.file}")
+    if not args.spec_file.exists():
+        raise FileNotFoundError(f"Spec file not found: {args.spec_file}")
+
+    # Validate the YAML early to fail fast if malformed.
+    with args.spec_file.open("r", encoding="utf-8") as handle:
+        yaml.safe_load(handle)
+
+    with h5py.File(args.file, "a") as handle:
+        _write_spec_snapshot(handle, args.spec_file, args.benchmark_name)
+        if args.kind == "calculation":
+            _write_run_metadata(
+                handle,
+                code_name=args.code_name,
+                code_version=args.code_version,
+                nuclear_data_name=args.nuclear_data_name,
+                nuclear_data_version=args.nuclear_data_version,
+                geometry=args.geometry,
+            )
+        else:
+            if "run_metadata" in handle:
+                del handle["run_metadata"]
+
+    print(f"Updated: {args.file}")
+
+
+if __name__ == "__main__":
+    main()

src/openmc_fusion_benchmarks/benchmark.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import yaml
+import h5py
 from pathlib import Path
 import warnings
 from abc import ABC, abstractmethod
@@ -127,6 +128,58 @@ def metadata(self):
 
         return self._metadata
 
+    def _write_spec_snapshot(self, filename: str = "benchmark_results.h5") -> None:
+        """Persist a snapshot of specifications.yaml into the results file."""
+        path = Path(filename)
+        if not path.exists():
+            warnings.warn(
+                f"Results file '{filename}' not found. Skipping spec snapshot.",
+                UserWarning,
+            )
+            return
+
+        spec_yaml = yaml.safe_dump(self._benchmark_spec, sort_keys=False)
+        spec_bytes = spec_yaml.encode("utf-8")
+
+        with h5py.File(path, "a") as handle:
+            if "specifications" in handle:
+                del handle["specifications"]
+            group = handle.create_group("specifications")
+            group.attrs["format"] = "yaml"
+            group.attrs["benchmark_name"] = self.name
+            group.create_dataset("yaml", data=np.bytes_(spec_bytes))
+
+    def _write_run_metadata(
+        self,
+        code_name: str,
+        code_version: str,
+        nuclear_data_name: str | None = None,
+        nuclear_data_version: str | None = None,
+        geometry: str | None = None,
+        filename: str = "benchmark_results.h5",
+    ) -> None:
+        """Persist run metadata into the results file."""
+        path = Path(filename)
+        if not path.exists():
+            warnings.warn(
+                f"Results file '{filename}' not found. Skipping run metadata.",
+                UserWarning,
+            )
+            return
+
+        with h5py.File(path, "a") as handle:
+            if "run_metadata" in handle:
+                del handle["run_metadata"]
+            group = handle.create_group("run_metadata")
+            group.attrs["code_name"] = str(code_name)
+            group.attrs["code_version"] = str(code_version)
+            if nuclear_data_name is not None:
+                group.attrs["nuclear_data_name"] = str(nuclear_data_name)
+            if nuclear_data_version is not None:
+                group.attrs["nuclear_data_version"] = str(nuclear_data_version)
+            if geometry is not None:
+                group.attrs["geometry"] = str(geometry)
+
 
 class OpenmcBenchmark(Benchmark):
     def __init__(self, name: str):
@@ -415,8 +468,12 @@ def _postprocess(self, statepoint: openmc.StatePoint | str | Path, mesh: str = '
         normalizer = make_default_openmc_normalizer(mesh)
 
         # Accept both already-open StatePoint objects and statepoint file paths.
+        code_version = "unknown"
+
         if isinstance(statepoint, openmc.StatePoint):
             sp = statepoint
+            if hasattr(sp, "version"):
+                code_version = ".".join(str(v) for v in sp.version)
             save_openmc_statepoint_tallies(
                 statepoint=sp,
                 filename="benchmark_results.h5",
@@ -428,6 +485,8 @@ def _postprocess(self, statepoint: openmc.StatePoint | str | Path, mesh: str = '
             )
         else:
             with openmc.StatePoint(str(statepoint)) as sp:
+                if hasattr(sp, "version"):
+                    code_version = ".".join(str(v) for v in sp.version)
                 save_openmc_statepoint_tallies(
                     statepoint=sp,
                     filename="benchmark_results.h5",
@@ -438,6 +497,18 @@ def _postprocess(self, statepoint: openmc.StatePoint | str | Path, mesh: str = '
                     normalizer=normalizer,
                 )
 
+        if hasattr(self, "_write_spec_snapshot"):
+            self._write_spec_snapshot("benchmark_results.h5")
+        if hasattr(self, "_write_run_metadata"):
+            self._write_run_metadata(
+                code_name="openmc",
+                code_version=code_version,
+                nuclear_data_name=None,
+                nuclear_data_version=None,
+                geometry="cad",
+                filename="benchmark_results.h5",
+            )
+
         return
 
     def _uncertainty_quantification(self, *args, **kwargs):

src/openmc_fusion_benchmarks/benchmark_results.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import yaml
 from pathlib import Path
 from typing import Union
 import h5py
@@ -163,6 +164,42 @@ def _loads_attr(attrs, key, default):
 
         return report
 
+    def get_spec_snapshot(self) -> dict:
+        """Return the embedded specifications.yaml snapshot if present."""
+        with h5py.File(self.filepath, "r") as f:
+            if "specifications" not in f:
+                raise ValueError("No specifications snapshot found in results file")
+            group = f["specifications"]
+            if "yaml" not in group:
+                raise ValueError("Specifications snapshot is missing 'yaml' dataset")
+            raw = group["yaml"][()]
+            if isinstance(raw, np.ndarray):
+                if raw.size == 0:
+                    raise ValueError("Specifications snapshot dataset is empty")
+                raw = raw[()]
+            if isinstance(raw, bytes):
+                yaml_text = raw.decode("utf-8")
+            else:
+                yaml_text = str(raw)
+        try:
+            return json.loads(json.dumps(yaml.safe_load(yaml_text)))
+        except Exception as exc:
+            raise ValueError("Failed to parse specifications snapshot YAML") from exc
+
+    def get_run_metadata(self) -> dict:
+        """Return run metadata embedded in the results file, if present."""
+        with h5py.File(self.filepath, "r") as f:
+            if "run_metadata" not in f:
+                raise ValueError("No run metadata found in results file")
+            attrs = f["run_metadata"].attrs
+            data: dict[str, str] = {}
+            for key, value in attrs.items():
+                if isinstance(value, bytes):
+                    data[key] = value.decode("utf-8")
+                else:
+                    data[key] = str(value)
+            return data
+
 
 class BenchmarkResults(Results):
     """Backward-compatible alias class for benchmark-centric naming."""

src/openmc_fusion_benchmarks/results_database/oktavian_al/experiment.h5

@@ -0,0 +1,111 @@
+import importlib.util
+import sys
+import types
+from pathlib import Path
+
+import h5py
+import pytest
+import yaml
+
+
+def _module_available(name: str) -> bool:
+    if name in sys.modules:
+        return True
+    try:
+        return importlib.util.find_spec(name) is not None
+    except ValueError:
+        return False
+
+
+# Provide minimal stubs so package imports work in minimal test envs.
+if not _module_available("openmc"):
+    openmc_stub = types.ModuleType("openmc")
+    openmc_stub.__path__ = []
+    sys.modules.setdefault("openmc", openmc_stub)
+
+if not _module_available("sandy"):
+    sys.modules.setdefault("sandy", types.ModuleType("sandy"))
+
+
+from scripts.backfill_results_metadata import main as backfill_main
+
+
+def _write_empty_results(filepath: Path) -> None:
+    with h5py.File(filepath, "w") as handle:
+        handle.create_group("tally_1")
+
+
+def test_backfill_experiment(tmp_path, monkeypatch):
+    results_path = tmp_path / "experiment.h5"
+    spec_path = tmp_path / "specifications.yaml"
+
+    _write_empty_results(results_path)
+    spec_path.write_text(yaml.safe_dump({"metadata": {"title": "Test"}}), encoding="utf-8")
+
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        [
+            "backfill_results_metadata.py",
+            "--file",
+            str(results_path),
+            "--spec-file",
+            str(spec_path),
+            "--kind",
+            "experiment",
+            "--benchmark-name",
+            "test",
+        ],
+    )
+
+    backfill_main()
+
+    with h5py.File(results_path, "r") as handle:
+        assert "specifications" in handle
+        assert "run_metadata" not in handle
+        spec_group = handle["specifications"]
+        assert spec_group.attrs["benchmark_name"] == "test"
+
+
+def test_backfill_calculation(tmp_path, monkeypatch):
+    results_path = tmp_path / "calc.h5"
+    spec_path = tmp_path / "specifications.yaml"
+
+    _write_empty_results(results_path)
+    spec_path.write_text(yaml.safe_dump({"metadata": {"title": "Test"}}), encoding="utf-8")
+
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        [
+            "backfill_results_metadata.py",
+            "--file",
+            str(results_path),
+            "--spec-file",
+            str(spec_path),
+            "--kind",
+            "calculation",
+            "--code-name",
+            "tripoli",
+            "--code-version",
+            "4.11",
+            "--nuclear-data-name",
+            "FENDL",
+            "--nuclear-data-version",
+            "3.2b",
+            "--geometry",
+            "cad",
+        ],
+    )
+
+    backfill_main()
+
+    with h5py.File(results_path, "r") as handle:
+        assert "specifications" in handle
+        assert "run_metadata" in handle
+        meta = handle["run_metadata"].attrs
+        assert meta["code_name"] == "tripoli"
+        assert meta["code_version"] == "4.11"
+        assert meta["nuclear_data_name"] == "FENDL"
+        assert meta["nuclear_data_version"] == "3.2b"
+        assert meta["geometry"] == "cad"