Reporting

Author: alongd

arc/level/reporting.py

@@ -42,10 +42,13 @@
 from typing import Any, Dict, List, Optional
 
 import nbformat
+import yaml
 from nbformat.v4 import new_code_cell, new_markdown_cell, new_notebook
 
+from arc.constants import E_h_kJmol
 from arc.exceptions import InputError
 from arc.level.protocol import CompositeProtocol
+from arc.parser.parser import parse_e_elect
 
 
 # =========================================================================== #
@@ -533,3 +536,151 @@ def _references_cell(sections: List[SpeciesSection]):
         for ref in ordered_refs:
             lines.append(f"- {ref}")
     return _md("\n".join(lines) + "\n", _cell_id("shared", "references"))
+
+
+# =========================================================================== #
+#  build_species_report_dict + write_species_report_yaml                      #
+# =========================================================================== #
+#
+# Per-species YAML report. Companion to the project-level provenance notebook
+# (which serves *independent verification* via Run-All); this writer produces
+# a consumable per-species summary that's readable in plain text and easy to
+# parse from downstream tooling. One file per stationary point, written by
+# the scheduler at composite finalization.
+
+
+def _composite_formula_text(protocol: CompositeProtocol) -> str:
+    """Plain-text version of the composite formula (no LaTeX).
+
+    The notebook uses LaTeX rendering; the YAML report is plain text — readers
+    of ``cat sp_composite_report.yml`` shouldn't see ``\\delta`` macros.
+    """
+    parts = [f"E_{protocol.base.label}"]
+    for term in protocol.corrections:
+        parts.append(term.label)
+    return "E_final = " + " + ".join(parts)
+
+
+def build_species_report_dict(
+    section: SpeciesSection,
+    e_elect_kj_per_mol: float,
+    timestamp: str,
+    arc_version: str,
+    arc_commit: str,
+) -> Dict[str, Any]:
+    """Assemble the per-species sp_composite report as a plain dict.
+
+    All energy values are computed by re-parsing the QM output files referenced
+    in ``section.sub_job_paths`` via :func:`arc.parser.parser.parse_e_elect`
+    (returns kJ/mol), then handed to each :class:`Term` to compute its
+    contribution. The same evaluation path the notebook's "Run All" follows —
+    the values land identically because the protocol is deterministic.
+
+    The caller-supplied ``e_elect_kj_per_mol`` is the value the scheduler
+    recorded on the ``ARCSpecies`` (set during ``_finalize_composite``). We
+    don't recompute the total here; we surface what ARC actually used so any
+    downstream tooling reading this report is consistent with the run's
+    output.yml / restart.yml.
+
+    Parameters
+    ----------
+    section : SpeciesSection
+        The reporting handoff struct populated by the scheduler at
+        finalization. Carries protocol, recipe, sub-job paths, flags.
+    e_elect_kj_per_mol : float
+        The final electronic energy ARC recorded for this species (kJ/mol).
+    timestamp : str
+        ISO-8601 string. Caller supplies for determinism (tests pin it).
+    arc_version, arc_commit : str
+        Provenance identifiers.
+
+    Returns
+    -------
+    dict
+        A plain dict ready for ``yaml.safe_dump`` or
+        :func:`write_species_report_yaml`.
+    """
+    energies_kj = {sl: parse_e_elect(p) for sl, p in section.sub_job_paths.items()}
+
+    base_term = section.protocol.base
+    base_sub_label, base_level = base_term.required_levels()[0]
+    base_block = {
+        "sub_label": base_sub_label,
+        "level": base_level.simple(),
+        "energy_kj_per_mol": energies_kj[base_sub_label],
+        "energy_hartree": energies_kj[base_sub_label] / E_h_kJmol,
+        "path": section.sub_job_paths[base_sub_label],
+    }
+
+    terms_block: List[Dict[str, Any]] = []
+    for term in section.protocol.corrections:
+        contribution_kj = term.evaluate(energies_kj)
+        sub_jobs: List[Dict[str, Any]] = []
+        for sub_label, level in term.required_levels():
+            sub_jobs.append({
+                "sub_label": sub_label,
+                "level": level.simple(),
+                "energy_kj_per_mol": energies_kj[sub_label],
+                "energy_hartree": energies_kj[sub_label] / E_h_kJmol,
+                "path": section.sub_job_paths[sub_label],
+            })
+        terms_block.append({
+            "label": term.label,
+            "type": type(term).__name__,
+            "contribution_kj_per_mol": contribution_kj,
+            "contribution_hartree": contribution_kj / E_h_kJmol,
+            "sub_jobs": sub_jobs,
+        })
+
+    return {
+        "species": section.label,
+        "kind": section.kind,
+        "generated_at": timestamp,
+        "arc_version": arc_version,
+        "arc_commit": arc_commit,
+        "protocol": {
+            "preset": section.preset_name,
+            "reference": section.reference,
+            "formula": _composite_formula_text(section.protocol),
+        },
+        "units": {
+            "energy": "kJ/mol",
+            "energy_alt": "Hartree",
+        },
+        "base": base_block,
+        "terms": terms_block,
+        "final": {
+            "e_elect_kj_per_mol": e_elect_kj_per_mol,
+            "e_elect_hartree": e_elect_kj_per_mol / E_h_kJmol,
+            "e_elect_source": "sp_composite",
+        },
+        "flags": list(section.flags),
+    }
+
+
+def write_species_report_yaml(
+    path: str,
+    section: SpeciesSection,
+    e_elect_kj_per_mol: float,
+    timestamp: str,
+    arc_version: str,
+    arc_commit: str,
+) -> None:
+    """Build and write the per-species sp_composite YAML report.
+
+    Creates the parent directory if missing. Output is deterministic: keys are
+    written in insertion order (Python 3.7+ dicts), flow style is block (the
+    YAML default), and ``sort_keys=False`` preserves the schema's natural
+    reading order (species → protocol → units → base → terms → final → flags).
+    Two writes with the same inputs produce byte-identical files.
+    """
+    report = build_species_report_dict(
+        section=section,
+        e_elect_kj_per_mol=e_elect_kj_per_mol,
+        timestamp=timestamp,
+        arc_version=arc_version,
+        arc_commit=arc_commit,
+    )
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+    with open(path, "w") as fh:
+        yaml.safe_dump(report, fh, sort_keys=False, default_flow_style=False)

arc/level/reporting_test.py

@@ -31,8 +31,10 @@
 from arc.level.protocol import CompositeProtocol
 from arc.level.reporting import (
     SpeciesSection,
+    build_species_report_dict,
     format_log_event,
     write_composite_notebook,
+    write_species_report_yaml,
 )
 
 
@@ -403,6 +405,199 @@ def test_notebook_executes_and_recomputes_expected_final_value(self):
         self.assertIn(f"{expected_kjmol:,.3f}", all_text)
 
 
+# --------------------------------------------------------------------------- #
+#  build_species_report_dict + write_species_report_yaml                      #
+# --------------------------------------------------------------------------- #
+
+
+import yaml  # noqa: E402  (import after the other module-level imports for grouping)
+
+
+class TestSpeciesReportDict(unittest.TestCase):
+    """``build_species_report_dict`` produces the consumable per-species summary.
+
+    The notebook (``sp_composite.ipynb``) is for *independent verification* via
+    Run-All; this YAML report is for *consumption* — readable in plain text,
+    parseable by tooling, one file per species with every term's contribution
+    spelled out next to the QM-output paths backing it.
+    """
+
+    def setUp(self):
+        self.tmp = tempfile.mkdtemp()
+        self.base_path = os.path.join(self.tmp, "base.out")
+        self.hi_path = os.path.join(self.tmp, "delta_T__high.out")
+        self.lo_path = os.path.join(self.tmp, "delta_T__low.out")
+        _write_gaussian_fixture(self.base_path, -76.345678)
+        _write_gaussian_fixture(self.hi_path, -76.346500)   # lower (more negative) → contribution = high - low < 0
+        _write_gaussian_fixture(self.lo_path, -76.345600)
+        self.section = _make_two_term_section(
+            paths={"base": self.base_path,
+                   "delta_T__high": self.hi_path,
+                   "delta_T__low": self.lo_path},
+        )
+
+    def tearDown(self):
+        # Tests create nested directories (e.g. Species/H2O/...) — use rmtree
+        # to clean them up wholesale rather than enumerating fixture files.
+        import shutil
+        shutil.rmtree(self.tmp, ignore_errors=True)
+
+    def test_top_level_fields(self):
+        d = build_species_report_dict(
+            section=self.section,
+            e_elect_kj_per_mol=-200000.0,
+            timestamp="2026-04-30T13:10:32Z",
+            arc_version="1.1.0",
+            arc_commit="74fc4fa5",
+        )
+        self.assertEqual(d["species"], "H2O")
+        self.assertEqual(d["kind"], "species")
+        self.assertEqual(d["generated_at"], "2026-04-30T13:10:32Z")
+        self.assertEqual(d["arc_version"], "1.1.0")
+        self.assertEqual(d["arc_commit"], "74fc4fa5")
+
+    def test_protocol_block(self):
+        d = build_species_report_dict(
+            section=self.section,
+            e_elect_kj_per_mol=-200000.0,
+            timestamp="2026-04-30T13:10:32Z",
+            arc_version="1.1.0",
+            arc_commit="abc",
+        )
+        self.assertIsNone(d["protocol"]["preset"])  # explicit recipe in fixture
+        self.assertIn("DOI", d["protocol"]["reference"])
+        # Formula spells out the sum the protocol evaluates.
+        self.assertIn("E_base", d["protocol"]["formula"])
+        self.assertIn("delta_T", d["protocol"]["formula"])
+
+    def test_units_block(self):
+        d = build_species_report_dict(
+            section=self.section,
+            e_elect_kj_per_mol=-200000.0,
+            timestamp="t",
+            arc_version="v",
+            arc_commit="c",
+        )
+        self.assertEqual(d["units"]["energy"], "kJ/mol")
+        self.assertEqual(d["units"]["energy_alt"], "Hartree")
+
+    def test_base_block(self):
+        d = build_species_report_dict(
+            section=self.section,
+            e_elect_kj_per_mol=-200000.0,
+            timestamp="t",
+            arc_version="v",
+            arc_commit="c",
+        )
+        base = d["base"]
+        self.assertEqual(base["sub_label"], "base")
+        self.assertEqual(base["path"], self.base_path)
+        # Energy parsed via arc.parser; cross-check Hartree↔kJ/mol consistency.
+        self.assertAlmostEqual(
+            base["energy_kj_per_mol"] / E_h_kJmol,
+            base["energy_hartree"],
+            places=6,
+        )
+
+    def test_terms_block_has_one_entry_per_correction(self):
+        d = build_species_report_dict(
+            section=self.section,
+            e_elect_kj_per_mol=-200000.0,
+            timestamp="t",
+            arc_version="v",
+            arc_commit="c",
+        )
+        self.assertEqual(len(d["terms"]), 1)  # fixture has exactly one correction
+        term = d["terms"][0]
+        self.assertEqual(term["label"], "delta_T")
+        self.assertEqual(term["type"], "DeltaTerm")
+        self.assertEqual(len(term["sub_jobs"]), 2)
+        self.assertEqual({sj["sub_label"] for sj in term["sub_jobs"]},
+                         {"delta_T__high", "delta_T__low"})
+        # Contribution = E[high] - E[low] = -76.346500 - (-76.345600) = -0.000900 Ha
+        # = -0.000900 × E_h_kJmol ≈ -2.363 kJ/mol
+        self.assertAlmostEqual(term["contribution_hartree"], -0.000900, places=6)
+        self.assertAlmostEqual(term["contribution_kj_per_mol"],
+                               -0.000900 * E_h_kJmol, places=3)
+
+    def test_final_block_uses_caller_supplied_e_elect(self):
+        d = build_species_report_dict(
+            section=self.section,
+            e_elect_kj_per_mol=-200000.123,
+            timestamp="t",
+            arc_version="v",
+            arc_commit="c",
+        )
+        self.assertEqual(d["final"]["e_elect_kj_per_mol"], -200000.123)
+        self.assertAlmostEqual(d["final"]["e_elect_hartree"],
+                               -200000.123 / E_h_kJmol, places=6)
+        self.assertEqual(d["final"]["e_elect_source"], "sp_composite")
+
+    def test_flags_propagated(self):
+        section = _make_two_term_section(
+            paths={"base": self.base_path,
+                   "delta_T__high": self.hi_path,
+                   "delta_T__low": self.lo_path},
+            flags=["MRCC degenerate-system fallback for delta_Q__high"],
+        )
+        d = build_species_report_dict(
+            section=section,
+            e_elect_kj_per_mol=-200000.0,
+            timestamp="t",
+            arc_version="v",
+            arc_commit="c",
+        )
+        self.assertEqual(len(d["flags"]), 1)
+        self.assertIn("MRCC", d["flags"][0])
+
+    def test_yaml_round_trips(self):
+        out = os.path.join(self.tmp, "sp_composite_report.yml")
+        write_species_report_yaml(
+            path=out,
+            section=self.section,
+            e_elect_kj_per_mol=-200000.0,
+            timestamp="2026-04-30T13:10:32Z",
+            arc_version="1.1.0",
+            arc_commit="74fc4fa5",
+        )
+        self.assertTrue(os.path.exists(out))
+        with open(out) as fh:
+            loaded = yaml.safe_load(fh)
+        self.assertEqual(loaded["species"], "H2O")
+        self.assertEqual(loaded["kind"], "species")
+        self.assertIn("base", loaded)
+        self.assertEqual(len(loaded["terms"]), 1)
+        self.assertEqual(loaded["terms"][0]["label"], "delta_T")
+
+    def test_yaml_writer_creates_parent_directory(self):
+        nested = os.path.join(self.tmp, "Species", "H2O", "sp_composite_report.yml")
+        write_species_report_yaml(
+            path=nested,
+            section=self.section,
+            e_elect_kj_per_mol=-200000.0,
+            timestamp="t",
+            arc_version="v",
+            arc_commit="c",
+        )
+        self.assertTrue(os.path.exists(nested))
+
+    def test_writer_is_deterministic(self):
+        """Two writes with the same inputs produce byte-identical files."""
+        out_a = os.path.join(self.tmp, "a.yml")
+        out_b = os.path.join(self.tmp, "b.yml")
+        for out in (out_a, out_b):
+            write_species_report_yaml(
+                path=out,
+                section=self.section,
+                e_elect_kj_per_mol=-200000.0,
+                timestamp="2026-04-30T13:10:32Z",
+                arc_version="1.1.0",
+                arc_commit="74fc4fa5",
+            )
+        with open(out_a, "rb") as fa, open(out_b, "rb") as fb:
+            self.assertEqual(fa.read(), fb.read())
+
+
 # --------------------------------------------------------------------------- #
 #  Import placement (module-level, per project guidelines)                    #
 # --------------------------------------------------------------------------- #