"""Resolve a recipe's stages into the calls vibe-qc would run, refusing the rest.
A recipe (:class:`vibesketch.document.model.Recipe`) is program-neutral. This
module spells each stage as the keywords of vibe-qc's ``run_job``, the way
:mod:`vibesketch.inputs.vibeqc` spells a job, so a reviewer sees exactly what
a launch would send, and refuses, with the reason, every option the route
cannot express (AGENTS.md, ground rule 5): a dispersion correction on the
molecular DFT and Hessian routes until vibe-qc propagates it (vibe-qc #548),
a functional or basis vibe-qc does not resolve, a MACE run without the
operator's licence acknowledgement, which a recipe never carries. Nothing is
translated into a different method.
Recipe containers target vibe-qc 0.17.25 or newer, which encodes the local
CCSD options record including the threshold preset and explicit triples mode.
The recipe's legacy ``t0`` spelling is refused: the engine calls that mode
``local``, and a requested method is never silently renamed.
The vocabulary comes from vibe-qc's source: ``method="mace"`` with
``MLIPOptions(model, device, dtype, weight_sha256, accept_academic_license)``;
``method="rks"`` with ``functional``, ``basis``, ``density_fit``, ``cosx``,
``grid_level``, ``rks_options`` (``conv_tol_energy``, ``conv_tol_grad``),
``optimize`` with ``optimizer_backend="ase"``, ``fmax`` and
``max_opt_steps``, and ``hessian`` with the finite-difference displacement;
``method="dlpno-ccsd(t)"`` with ``dlpno_thresholds`` (``loose``, ``normal``,
``tight``), ``frozen_core`` (``chemical`` or ``all-electron``) and
``dlpno_ccsd_options={"triples_mode": ...}``.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
from vibesketch.document.model import Recipe, Species, Stage
from vibesketch.inputs.vibeqc import _DISPERSION_INCLUDED, _FUNCTIONALS, _basis
__all__ = ["ResolvedStage", "resolve_recipe", "resolve_stage"]
#: vibe-qc's preset names for the published PNO threshold sets.
_THRESHOLDS = {"LoosePNO": "loose", "NormalPNO": "normal", "TightPNO": "tight"}
[docs]
@dataclass(frozen=True)
class ResolvedStage:
"""What one stage of a recipe would run, or why it cannot."""
stage: str
kind: str
program: str
#: The ``run_job`` keywords, in vibe-qc's names; empty when refused.
keywords: dict[str, Any] = field(default_factory=dict)
#: Settings the operator supplies at launch, never saved in the recipe.
at_launch: dict[str, str] = field(default_factory=dict)
#: What the stage's validator checks on the result, in the recipe's units.
checks: dict[str, Any] = field(default_factory=dict)
refusals: list[str] = field(default_factory=list)
[docs]
@property
def refused(self) -> bool:
return bool(self.refusals)
[docs]
def to_dict(self) -> dict[str, Any]:
return {
"stage": self.stage,
"kind": self.kind,
"program": self.program,
"keywords": dict(self.keywords),
"at_launch": dict(self.at_launch),
"checks": dict(self.checks),
"refusals": list(self.refusals),
}
def _surface_keywords(surface: Any, refusals: list[str]) -> dict[str, Any]:
functional = _FUNCTIONALS.get(surface.functional.lower())
if functional is None:
refusals.append(
f"vibe-qc's XC table does not resolve the functional {surface.functional!r}; "
"it is not swapped for another"
)
basis = _basis(surface.basis)
if basis is None:
refusals.append(f"vibe-qc ships no basis set {surface.basis!r}")
if surface.dispersion is not None:
if functional in _DISPERSION_INCLUDED:
refusals.append(
f"{surface.functional} carries its own dispersion; a {surface.dispersion} "
"correction on top is not expressible"
)
else:
refusals.append(
f"a {surface.dispersion} correction is not expressible on vibe-qc's molecular "
"DFT and finite-difference Hessian routes until vibe-qc #548 propagates it; "
"the publication baseline runs without dispersion"
)
return {
"method": "rks",
"functional": functional,
"basis": basis,
"density_fit": surface.density_fit,
"cosx": surface.cosx,
"grid_level": surface.grid,
"dispersion": None,
"rks_options": {
"conv_tol_energy": surface.scf_energy_tolerance,
"conv_tol_grad": surface.scf_gradient_tolerance,
**({"max_iter": surface.scf_max_iterations} if surface.scf_max_iterations else {}),
**(
{"stability_davidson_max_iter": surface.stability_max_iterations}
if surface.stability_max_iterations
else {}
),
},
}
[docs]
def resolve_stage(stage: Stage, species: Species | None = None) -> ResolvedStage:
"""``stage`` as vibe-qc's ``run_job`` keywords, or the reasons it is refused.
``species`` adds the charge and multiplicity vibe-qc's ``Molecule`` takes;
a species with an unpaired electron makes the Kohn-Sham route ``uks``.
"""
refusals: list[str] = []
keywords: dict[str, Any] = {}
at_launch: dict[str, str] = {}
checks: dict[str, Any] = {}
if species is not None:
keywords["charge"] = species.charge
keywords["multiplicity"] = species.multiplicity
if stage.kind == "calculation":
return ResolvedStage(
stage.id,
stage.kind,
stage.calculation.program if stage.calculation else "unknown",
refusals=[
"ordinary writer stage: use compile_energy_recipe, then the recorded job "
"submission/result-import path; the publication campaign driver does not "
"execute this kind"
],
)
if stage.kind == "prepare":
return ResolvedStage(stage.id, stage.kind, "vibe-sketch", {"method": "etkdg+mmff94"})
if stage.kind == "mace" and stage.mace is not None:
options = stage.mace
keywords |= {
"method": "mace",
"optimize": True,
"optimizer_backend": "ase",
"fmax": options.fmax_ev_per_angstrom,
"max_opt_steps": options.max_steps,
"mlip_options": {
"model": options.model,
"device": options.device,
"dtype": options.dtype,
"weight_sha256": options.weight_sha256,
},
}
at_launch["accept_academic_license"] = (
"an eligible operator acknowledges the model's licence when launching; a recipe "
"never carries the acknowledgement"
)
checks = {"fmax_ev_per_angstrom": options.fmax_ev_per_angstrom}
elif stage.kind == "dft" and stage.dft is not None:
keywords |= _surface_keywords(stage.dft.surface, refusals)
keywords |= {
"optimize": True,
"optimizer_backend": stage.dft.optimizer,
"fmax": stage.dft.fmax_ev_per_angstrom,
"max_opt_steps": stage.dft.max_steps,
}
checks = {"fmax_ev_per_angstrom": stage.dft.fmax_ev_per_angstrom}
elif stage.kind == "hessian" and stage.hessian is not None:
keywords |= _surface_keywords(stage.hessian.surface, refusals)
keywords["hessian"] = True
# This is an executed setting, not merely a validator annotation (#150).
# Older container runners must refuse this unknown option.
keywords["hessian_step_bohr"] = stage.hessian.displacement_bohr
checks = {
"displacement_bohr": stage.hessian.displacement_bohr,
"imaginary_mode_threshold_cm": stage.hessian.imaginary_mode_threshold_cm,
}
if stage.hessian.require_positive_modes:
checks["require_positive_modes"] = True
if stage.hessian.compare_parent_max_delta_cm is not None:
checks["compare_parent_max_delta_cm"] = stage.hessian.compare_parent_max_delta_cm
checks["comparison_stage"] = stage.parent
elif stage.kind == "dlpno" and stage.dlpno is not None:
dlpno = stage.dlpno
basis = _basis(dlpno.basis)
if basis is None:
refusals.append(f"vibe-qc ships no basis set {dlpno.basis!r}")
aux = None
if dlpno.aux_basis is not None:
aux = _basis(dlpno.aux_basis)
if dlpno.aux_basis.lower() == "cc-pvdz-rifit":
aux = "cc-pvdz-rifit" # shipped QC fitting basis for this endpoint
if aux is None:
refusals.append(f"vibe-qc ships no fitting basis {dlpno.aux_basis!r}")
if species is not None and species.multiplicity != 1:
refusals.append(
"the DLPNO stage is specified for a closed-shell reference; "
f"{species.id} has multiplicity {species.multiplicity}"
)
if dlpno.reference_contract == "real_restricted_cc":
keywords["reference_contract"] = dlpno.reference_contract
checks["reference_contract"] = dlpno.reference_contract
if dlpno.triples_mode != "t1":
refusals.append(
"vibe-qc has no triples_mode='t0'; its T0 mode is named 'local'. "
"Select the supported recipe mode 't1'; the requested mode is not renamed"
)
keywords |= {
"method": dlpno.method,
"basis": basis,
"aux_basis": aux,
"frozen_core": dlpno.frozen_core,
"dlpno_ccsd_options": {
"dlpno_thresholds": _THRESHOLDS[dlpno.thresholds],
"triples_mode": dlpno.triples_mode,
},
}
rhf = {
key: value
for key, value in {
"max_iter": dlpno.scf_max_iterations,
"stability_davidson_max_iter": dlpno.stability_max_iterations,
"conv_tol_energy": dlpno.scf_energy_tolerance,
"conv_tol_grad": dlpno.scf_gradient_tolerance,
}.items()
if value is not None
}
if rhf:
keywords["rhf_options"] = rhf
keywords["dlpno_ccsd_options"].update(
{
key: value
for key, value in {
"max_iter": dlpno.cc_max_iterations,
"conv_tol_energy": dlpno.cc_energy_tolerance,
"conv_tol_residual": dlpno.cc_residual_tolerance,
"aux_basis": aux,
}.items()
if value is not None
}
)
at_launch["resolved_pno_thresholds"] = (
"Inspect and retain every numeric threshold from the selected runtime's "
"resolved LocalCCSDOptions; a preset name alone is not numerical provenance."
)
else: # unreachable for a validated stage
refusals.append(f"stage {stage.id} has no options for its kind {stage.kind!r}")
if species is not None and species.multiplicity != 1 and keywords.get("method") == "rks":
keywords["method"] = "uks"
keywords["uks_options"] = keywords.pop("rks_options")
return ResolvedStage(
stage.id,
stage.kind,
"vibeqc",
{} if refusals else keywords,
at_launch,
checks,
refusals,
)
[docs]
def resolve_recipe(recipe: Recipe, species: dict[str, Species]) -> dict[str, list[ResolvedStage]]:
"""Every stage of ``recipe`` resolved per species id, in stage order."""
return {
identifier: [resolve_stage(stage, species[identifier]) for stage in recipe.stages]
for identifier in recipe.species
if identifier in species
}