"""QVF, the vibyverse's file format, as vibe-sketch's document format.
Every program of the vibe family keeps its files as QVF (the qvf
repository's spec/qvf-format-spec.md, toolkit release 0.6). A vibe-sketch QVF holds:
- a vendor section ``x_vibesketch.document``, not critical, its namespace
declared in the manifest's ``extensions``, with the members ``document``
(the vibe-sketch document, schema ``vibe-sketch/1``, drawing included),
``species`` (each species' identifiers, including those without a 3D
structure) and ``sdf`` (the species as an SD file);
- a ``preview`` (spec § 5.15): a PNG of the species as the thumbnail, at most
1024 pixels on its longest side, and the scheme as SVG;
- one ``structure`` section per species that has a 3D conformer, named
``structure_<species id>``, with its bonds and its identity (§ 5.1):
charge, multiplicity, SMILES, InChI, InChIKey, a checked name and the
compound number as ``label``;
- a ``reaction.scheme`` (§ 5.14) with the steps whose reactants and products
all have a structure section;
- a ``job.spec`` (§ 5.9) for each calculation on such a species, a job set
whose jobs name their structure.
So any QVF reader shows the molecules, knows which is which and what the
scheme and its calculations are, without knowing vibe-sketch.
A vibe-qc calculation is submitted as a **QVF job container** (§ 5.9,
:func:`job_container`): one ``structure``, one ``job.spec`` in the names
vibe-qc's ``run_job`` takes, the ``x_vibesketch.document`` vendor section
with the whole document, and ``provenance.run_status: "pending"``.
``vibeqc run job.qvf`` updates that file in place, so the result keeps the
drawing that asked for it.
Reading takes the document from ``x_vibesketch.document`` when there is
one. A finished container (``run_status`` stated) also gives back its
result: the runner's structure becomes a new conformer of the job's
species, and :attr:`~vibesketch.document.derive.Derivation.result` says
what happened: ``job``, ``species``, ``status`` (``pending``, ``running``,
``converged`` or ``failed``), ``program``, ``program_version``, ``method``,
``tasks``, ``energy`` (``provenance.scf_energy`` as written, ``{value,
units}``), ``energy_hartree`` (the same in hartree when the units say so),
``conformer`` (the index of the new conformer, or ``None``), ``exit_status``,
``started_utc``, ``finished_utc`` and ``wall_seconds``. A QVF from another
program, such as a vibe-qc result, contributes its ``structure`` sections as
species: bonds perceived from the 3D coordinates with the charge the section
states (or its SMILES implies), a 2D layout for the drawing, the coordinates
kept as the conformer, and the section's multiplicity, label and name.
qvf-writer, which provides the ``qvf_writer`` and ``qvf_reader`` modules, is
an optional dependency installed from the qvf checkout.
"""
from __future__ import annotations
import contextlib
import hashlib
import io
import json
import re
import zipfile
from typing import Any
from vibesketch.document.derive import Derivation
from vibesketch.document.model import Conformer, Document, Job, Species
from vibesketch.document.results import Selection
__all__ = [
"stage_container",
"DOCUMENT_KIND",
"MEDIA_TYPE",
"NAMESPACE",
"QvfError",
"QvfUnavailable",
"from_qvf",
"job_container",
"overwrite_refusal",
"qvf_available",
"species_qvf",
"to_qvf",
]
NAMESPACE = "x_vibesketch"
DOCUMENT_KIND = "x_vibesketch.document"
#: The version of the ``x_vibesketch`` extension this module writes.
EXTENSION_VERSION = "1"
#: The qvf release vibe-sketch is pinned to: the ``qvf`` extra's floor and CI's
#: ``QVF_TAG`` name the same one (tests/test_qvf.py holds them together).
QVF_WRITER_MINIMUM = (0, 6)
#: QVF's registered media type (the qvf repository's spec/file-type).
MEDIA_TYPE = "application/vnd.vibe-qc.qvf+zip"
[docs]
class QvfError(ValueError):
"""A QVF that vibe-sketch cannot read, or cannot write."""
[docs]
class QvfUnavailable(QvfError):
"""qvf-writer is not installed, or is older than the supported toolkit release."""
def __init__(self, found: str = "QVF needs qvf-writer") -> None:
super().__init__(
f"{found}: install it from a qvf checkout with ./scripts/install.sh "
"--qvf-path <qvf checkout>, or pip install '<qvf checkout>/python[numpy,validate]'"
)
def _writer_version() -> tuple[int, ...] | None:
"""The installed qvf-writer's release as numbers, or ``None`` when it does not say."""
from importlib import metadata
try:
release = metadata.version("qvf-writer")
except metadata.PackageNotFoundError:
return None
numbers = re.findall(r"\d+", release)
return tuple(int(n) for n in numbers[:3])
def _modules() -> tuple[Any, Any]:
try:
import qvf_reader
import qvf_writer
except ImportError:
raise QvfUnavailable() from None
found = _writer_version()
if found is not None and found < QVF_WRITER_MINIMUM:
wanted = ".".join(map(str, QVF_WRITER_MINIMUM))
have = ".".join(map(str, found))
raise QvfUnavailable(f"QVF needs qvf-writer {wanted} or newer, not {have}")
return qvf_writer, qvf_reader
[docs]
def qvf_available() -> bool:
"""Whether qvf-writer is installed, so QVF can be read and written."""
try:
_modules()
except QvfUnavailable:
return False
return True
# ── Writing ─────────────────────────────────────────────────────────────
def _bonds(species: Species, conformer: Conformer) -> list[dict[str, Any]]:
"""The bonds between the conformer's atoms, ``{i, j, order}``.
A conformer vibe-sketch embedded has the species' atoms in its own order
(the drawing's atoms, then the hydrogens), so the bonds come from the
structure itself, and a generated nanostructure records its own; for any
other conformer they are perceived from the coordinates.
"""
if species.nanostructure is not None and conformer.symbols == species.conformers[0].symbols:
return [{"i": i, "j": j, "order": order} for i, j, order in species.nanostructure.bonds]
from rdkit import Chem
from vibesketch.chem.species import mol_from_smiles
if species.smiles is not None:
with_h = Chem.AddHs(mol_from_smiles(species.smiles))
if [a.GetSymbol() for a in with_h.GetAtoms()] == conformer.symbols:
return [
{
"i": b.GetBeginAtomIdx(),
"j": b.GetEndAtomIdx(),
"order": float(b.GetBondTypeAsDouble()),
}
for b in with_h.GetBonds()
]
from vibesketch.chem.perceive import perceive
template = mol_from_smiles(species.smiles) if species.smiles is not None else None
found = perceive(
conformer.symbols, conformer.xyz_angstrom, charge=species.charge, template=template
)
if found is None:
return []
return [
{"i": b.GetBeginAtomIdx(), "j": b.GetEndAtomIdx(), "order": float(b.GetBondTypeAsDouble())}
for b in found.mol.GetBonds()
]
def _what(species: Species) -> str:
"""How a calculation label names a species: its name, SMILES or nanostructure."""
if species.name is not None:
return species.name.value
if species.nanostructure is not None:
return species.nanostructure.description
return species.smiles or species.id
def _atoms(conformer: Conformer) -> list[dict[str, Any]]:
return [
{"symbol": symbol, "position": position}
for symbol, position in zip(conformer.symbols, conformer.xyz_angstrom, strict=True)
]
def _inchi(species: Species) -> str | None:
from vibesketch.chem.species import mol_from_smiles, properties
if species.smiles is None:
return None
inchi = properties(mol_from_smiles(species.smiles)).inchi
return inchi if inchi and inchi.startswith("InChI=") else None
def _identity(species: Species) -> dict[str, Any]:
"""A species' identity fields for its structure section (spec § 5.1).
Only a name the user gave, or one checked as well, is written: the QVF
spec asks for identifiers known to describe the atoms.
"""
fields: dict[str, Any] = {"charge": species.charge, "multiplicity": species.multiplicity}
if species.label is not None:
fields["label"] = species.label
if species.name is not None and species.name.confidence == "high":
fields["name"] = species.name.value
if species.smiles is not None:
fields["smiles"] = species.smiles
inchi = _inchi(species)
if inchi is not None:
fields["inchi"] = inchi
if species.inchikey is not None:
fields["inchikey"] = species.inchikey
return fields
def _species_index(document: Document) -> list[dict[str, Any]]:
"""Each species with its identifiers, those without a 3D structure included."""
from vibesketch.chem.species import mol_from_smiles, properties
index = []
for species in document.species:
entry: dict[str, Any] = {"id": species.id}
if species.label is not None:
entry["label"] = species.label
if species.name is not None:
entry["name"] = species.name.value
if species.smiles is not None:
facts = properties(mol_from_smiles(species.smiles))
entry |= {"smiles": species.smiles, "inchi": facts.inchi, "formula": facts.formula}
entry |= {
"inchikey": species.inchikey,
"charge": species.charge,
"multiplicity": species.multiplicity,
"structure": f"structure_{species.id}" if species.conformers else None,
}
index.append(entry)
return index
def _sdf(document: Document) -> str | None:
from vibesketch.document.formats import ExportRefused, export
try:
return export(document, "sdf").content.decode("utf-8")
except ExportRefused:
return None
def _add_preview(writer: Any, document: Document) -> None:
"""The ``preview`` section, when the drawing libraries are there to draw it."""
from vibesketch.render import RenderRefused, RenderUnavailable, render_svg, thumbnail
try:
png, _, _ = thumbnail(document)
except (RenderRefused, RenderUnavailable):
return
svg = None
with contextlib.suppress(RenderRefused, RenderUnavailable):
svg = render_svg(document)
writer.add_preview(
png,
thumbnail_media_type="image/png",
image=svg,
image_media_type="image/svg+xml" if svg is not None else None,
)
def _yield_percent(text: str) -> float | None:
"""The number of a yield printed as ``87%`` or ``87.5 %``; ``None`` for ``quant.`` and such."""
match = re.fullmatch(r"\s*(\d+(?:[.,]\d+)?)\s*%\s*", text)
if match is None:
return None
value = float(match.group(1).replace(",", "."))
return value if 0 <= value <= 100 else None
def _steps(document: Document, structures: set[str]) -> list[dict[str, Any]]:
"""The reaction steps whose reactants and products all have a structure section.
vibe-sketch records a step's species by their place at the arrow; QVF by
their chemical role (spec § 5.14). A retrosynthetic arrow points from the
target to its precursors, so the species at its tail are the step's
products. A reagent without a 3D structure is named in the conditions.
"""
steps = []
for reaction in document.reactions:
tail, head = reaction.reactants, reaction.products
if not set(tail + head) <= structures:
continue
reactants, products = (head, tail) if reaction.arrow == "retrosynthetic" else (tail, head)
step: dict[str, Any] = {
"id": reaction.id,
"reactants": [{"structure_ref": f"structure_{s}"} for s in reactants],
"products": [{"structure_ref": f"structure_{s}"} for s in products],
}
if reaction.arrow != "forward":
step["arrow"] = reaction.arrow
if reaction.arrow == "resonance": # no reaction: no reagents, conditions or yield
steps.append(step)
continue
drawn = [s for s in reaction.reagents if s in structures]
if drawn:
step["reagents"] = [{"structure_ref": f"structure_{s}"} for s in drawn]
named = []
for sid in reaction.reagents:
if sid not in structures:
reagent = document.species_by_id(sid)
named.append(reagent.name.value if reagent.name else reagent.smiles or sid)
if reaction.conditions or named:
step["conditions"] = [*named, *reaction.conditions]
if reaction.yield_ is not None:
step["yield_text"] = reaction.yield_
percent = _yield_percent(reaction.yield_)
if percent is not None:
step["yield_percent"] = percent
steps.append(step)
return steps
#: vibe-sketch's tasks as QVF ``job.spec`` task lists (vibe-qc's runner knows the first three).
_TASKS = {
"sp": ["single_point"], "opt": ["optimize"], "freq": ["hessian"],
"opt+freq": ["optimize", "hessian"], "tddft": ["excited_states"], "nmr": ["nmr"],
} # fmt: skip
def _job_spec(job: Job, species: Species) -> dict[str, Any] | None:
"""``add_job_spec``'s arguments for a job, or ``None`` for one that computes nothing.
The typed fields carry what every program reads: method and functional
in vibe-qc's names, basis, dispersion, solvation, resources, the
species' charge and multiplicity. What only some programs can do (an
auxiliary basis, an approximation such as RIJCOSX, excited states,
constraints) goes into ``options``, which a runner that does not know an
option must refuse (spec § 5.9): the job fails closed, as the writers do.
"""
from vibesketch.inputs.vibeqc import method_names
spec = job.spec
if spec.task == "geometry" or spec.method is None:
return None
names = method_names(spec.method, species.multiplicity)
method, functional = names if names is not None else (spec.method.lower(), None)
fields: dict[str, Any] = {
"job_type": "molecular",
"method": method,
"charge": species.charge,
"multiplicity": species.multiplicity,
"tasks": _TASKS[spec.task],
"structure_ref": f"structure_{species.id}",
"label": f"{job.id}: {spec.method} {spec.task} for {job.program}",
}
if functional is not None:
fields["functional"] = functional
if spec.basis is not None:
fields["basis"] = spec.basis.lower()
if spec.dispersion is not None:
fields["dispersion"] = re.sub(r"[()]", "", spec.dispersion.lower())
if spec.solvent is not None:
solvent = spec.solvent
fields["solvation"] = {"model": solvent.model.lower(), "solvent": solvent.name.lower()}
if spec.resources is not None:
resources: dict[str, Any] = {"n_cores": spec.resources.nprocs}
if spec.resources.maxcore_mb is not None:
resources["memory_mb"] = spec.resources.nprocs * spec.resources.maxcore_mb
fields["resources"] = resources
options: dict[str, Any] = {}
if spec.aux_basis is not None:
options["aux_basis"] = spec.aux_basis.lower()
if spec.approximation is not None:
options["approximation"] = spec.approximation.lower()
if spec.states is not None:
options["states"] = spec.states
if spec.constraints:
options["constraints"] = [c.model_dump(exclude_none=True) for c in spec.constraints]
if options:
fields["options"] = options
return fields
[docs]
def to_qvf(
document: Document, *, preview: bool = True, result_selection: Selection | None = None
) -> bytes:
"""The document as a QVF archive; raises :class:`QvfUnavailable`."""
from vibesketch.document.results import reaction_energies
rows = reaction_energies(document, result_selection) if result_selection is not None else None
return _to_qvf_rows(document, preview=preview, rows=rows)
def _to_qvf_rows(
document: Document, *, preview: bool = True, rows: list[dict[str, Any]] | None = None
) -> bytes:
"""Package already-validated reports in the existing reaction-scheme extension."""
from vibesketch import __version__
qvf_writer, _ = _modules()
writer = qvf_writer.QvfWriter(program="vibe-sketch", version=__version__, calculation="")
writer.set_extensions({NAMESPACE: {"version": EXTENSION_VERSION, "critical": False}})
members: dict[str, Any] = {"document": document.to_json_data()}
if document.species:
members["species"] = {"species": _species_index(document)}
sdf = _sdf(document)
if sdf is not None:
members["sdf"] = {"media_type": "chemical/x-mdl-sdfile", "text": sdf}
writer.add_vendor_section(
DOCUMENT_KIND,
json_members=members,
label="vibe-sketch document",
section_id="vibesketch_document",
)
if preview and document.species:
_add_preview(writer, document)
structures: set[str] = set()
for species in document.species:
if not species.conformers:
continue
conformer = species.conformers[0]
writer.add_structure(
_atoms(conformer),
bonds=_bonds(species, conformer),
section_id=f"structure_{species.id}",
**_identity(species),
)
structures.add(species.id)
steps = _steps(document, structures)
if rows is not None:
reports = {r["reaction"]: r for r in rows}
for step in steps:
if step["id"] in reports:
report = reports[step["id"]]
step["x_vibesketch"] = {"energies": report}
for role in ("reactants", "products"):
for participant in step[role]:
sid = participant["structure_ref"].removeprefix("structure_")
participant["coefficient"] = report["coefficients"][role][sid]
if steps:
writer.add_reaction_scheme(steps, label="reaction scheme")
for job in document.jobs:
if job.species in structures:
fields = _job_spec(job, document.species_by_id(job.species))
if fields is not None:
writer.add_job_spec(**fields, section_id=f"job_{job.id}")
return bytes(writer.to_bytes())
[docs]
def overwrite_refusal(data: bytes) -> str | None:
"""Why replacing this archive would lose data, or ``None`` for a native sketch.
Results are imported as structures, not retained as opaque payloads. Only
archives reproducible from their stored sketch may be overwritten. Preview
images are allowed to change; calculation data, additional metadata and
unknown members require Save As. Invalid archives fail closed too.
"""
from vibesketch.limits import LimitExceeded, check_archive
advice = "Use Save As with a different destination to keep the original QVF intact."
try:
check_archive(data)
except LimitExceeded as error:
return f"{error}. {advice}"
try:
with zipfile.ZipFile(io.BytesIO(data)) as original:
manifest = json.loads(original.read("manifest.json"))
sections = manifest["sections"]
if not any(s["kind"] == DOCUMENT_KIND for s in sections):
return f"This QVF is an imported structure or result, not a native sketch. {advice}"
kinds = {DOCUMENT_KIND, "preview", "structure", "reaction.scheme", "job.spec"}
extra = sorted({s["kind"] for s in sections} - kinds)
if extra:
kinds_text = ", ".join(extra)
return (
f"This QVF contains data sketch export cannot retain ({kinds_text}). {advice}"
)
# Compare the stored native document's reproducible payloads, so
# extra data within otherwise familiar section kinds is protected.
document = from_qvf(data).document
with zipfile.ZipFile(io.BytesIO(to_qvf(document, preview=False))) as rebuilt:
expected = json.loads(rebuilt.read("manifest.json"))
for key in ("qvf_version", "schema_uri", "byte_order", "extensions"):
if manifest.get(key) != expected.get(key):
return f"This QVF has unsupported {key} metadata. {advice}"
if (
set(manifest) != set(expected)
or set(manifest["source"]) != set(expected["source"])
or manifest["source"]["program"] != "vibe-sketch"
or manifest["source"]["calculation"] != ""
or set(manifest["metadata"]) != set(expected["metadata"])
):
return f"This QVF has additional provenance or metadata. {advice}"
actual = _replaceable_sections(original, sections, preview=True)
wanted = _replaceable_sections(rebuilt, expected["sections"], preview=False)
if actual != wanted:
return f"This QVF contains payloads not reproduced by its sketch. {advice}"
except Exception:
# Inspection is a deletion guard over untrusted archives, including
# toolkit/codec failures. Never turn an inspection failure into consent.
return f"Cannot verify that this QVF can be replaced without data loss. {advice}"
return None
def _replaceable_sections(
archive: zipfile.ZipFile, sections: list[dict[str, Any]], *, preview: bool
) -> dict[str, Any]:
"""A semantic inventory, verifying every member and rejecting unlisted data."""
inventory: dict[str, Any] = {}
paths = {"manifest.json"}
identifiers: set[str] = set()
for section in sections:
identifier = section["id"]
if identifier in identifiers:
raise ValueError("duplicate section")
identifiers.add(identifier)
is_preview = preview and section["kind"] == "preview"
members = section["members"]
if is_preview and (
identifier != "preview"
or set(section) != {"id", "kind", "members"}
or not {"thumbnail"} <= members.keys() <= {"thumbnail", "image"}
):
raise ValueError("additional preview data")
values: dict[str, Any] = {}
for name, descriptor in members.items():
path = descriptor["path"]
if path in paths:
raise ValueError("duplicate member")
paths.add(path)
content = archive.read(path)
if (
descriptor["size"] != len(content)
or descriptor["sha256"] != hashlib.sha256(content).hexdigest()
):
raise ValueError("damaged member")
metadata = {k: v for k, v in descriptor.items() if k not in {"path", "size", "sha256"}}
if is_preview:
media = "image/png" if name == "thumbnail" else "image/svg+xml"
if metadata != {"format": "binary", "media_type": media}:
raise ValueError("unknown preview member")
else:
values[name] = (metadata, json.loads(content))
if not is_preview:
inventory[identifier] = (
{k: v for k, v in section.items() if k != "members"},
values,
)
if len(archive.namelist()) != len(paths) or set(archive.namelist()) != paths:
raise ValueError("unlisted archive entries")
return inventory
[docs]
def species_qvf(document: Document, species_id: str, *, document_id: str | None = None) -> bytes:
"""One species as a QVF of its own: one structure with its identity.
This is the file a 3D viewer opens. The charge and multiplicity are on
the structure section and, as the spec asks of a single structure, in
``provenance`` too. Raises ``KeyError`` for an unknown species,
:class:`QvfError` for one without a conformer.
"""
from vibesketch import __version__
qvf_writer, _ = _modules()
species = document.species_by_id(species_id)
if species.solid is not None:
raise QvfError(
f"species {species_id} is the solid {species.solid.text}, given by its formula "
"without a structure (roadmap Phase 5)"
)
if not species.conformers:
raise QvfError(f"species {species_id} has no 3D structure yet")
conformer = species.conformers[0]
what = _what(species)
writer = qvf_writer.QvfWriter(
program="vibe-sketch", version=__version__, calculation=f"{species_id}: {what}"
)
writer.set_provenance(charge=species.charge, multiplicity=species.multiplicity)
writer.add_structure(_atoms(conformer), bonds=_bonds(species, conformer), **_identity(species))
from vibesketch.document.geometry_return import origin
writer.set_extensions({NAMESPACE: {"version": EXTENSION_VERSION, "critical": False}})
writer.add_vendor_section(
"x_vibesketch.origin",
json_members={"origin": origin(document, species_id, document_id)},
section_id="vibesketch_origin",
)
return bytes(writer.to_bytes())
[docs]
def job_container(document: Document, job: Job, *, preview: bool = True) -> bytes:
"""A pending QVF job container for one vibe-qc job: what ``vibeqc run`` executes.
The archive holds the job's species as its one ``structure`` (the first
conformer, with its identity), one ``job.spec`` in ``run_job``'s names
(the ``vibeqc`` writer's translation, which refuses what it cannot
express), the whole ``document`` in the ``x_vibesketch.document`` vendor
section, a preview of the scheme, and ``provenance.run_status:
"pending"``. The runner updates the same file in place and keeps the
vendor section, so the result opens back with its drawing
(:func:`from_qvf`). ``document`` should already list ``job``, so the
result names it. Raises ``KeyError`` for an unknown species,
:class:`QvfError` for one without a conformer, the writer's
``WriterRefusal`` for a specification it cannot express, and
:class:`QvfUnavailable` without qvf-writer.
"""
from vibesketch import __version__
from vibesketch.inputs.vibeqc import VibeqcWriter
qvf_writer, _ = _modules()
species = document.species_by_id(job.species)
if not species.conformers:
raise QvfError(f"species {job.species} has no 3D structure yet")
fields = VibeqcWriter().container_spec(species, job.spec)
conformer = species.conformers[0]
what = _what(species)
writer = qvf_writer.QvfWriter(
program="vibe-sketch",
version=__version__,
calculation=f"{job.id}: {job.spec.method} {job.spec.task} for {job.species}: {what}",
)
writer.set_extensions({NAMESPACE: {"version": EXTENSION_VERSION, "critical": False}})
members: dict[str, Any] = {
"document": document.to_json_data(),
"species": {"species": _species_index(document)},
}
sdf = _sdf(document)
if sdf is not None:
members["sdf"] = {"media_type": "chemical/x-mdl-sdfile", "text": sdf}
writer.add_vendor_section(
DOCUMENT_KIND,
json_members=members,
label="vibe-sketch document",
section_id="vibesketch_document",
)
if preview:
_add_preview(writer, document)
structure = writer.add_structure(
_atoms(conformer),
bonds=_bonds(species, conformer),
section_id=f"structure_{species.id}",
**_identity(species),
)
writer.add_job_spec(
**fields,
structure_ref=structure,
label=f"{job.id}: {job.spec.method} {job.spec.task} for vibeqc",
section_id=f"job_{job.id}",
)
writer.set_provenance(
run_status="pending", charge=species.charge, multiplicity=species.multiplicity
)
return bytes(writer.to_bytes())
#: The container runner's tasks per recipe stage kind.
_STAGE_TASKS = {
"mace": ["optimize"],
"dft": ["optimize"],
"hessian": ["hessian"],
"dlpno": ["single_point"],
}
#: Resolved keywords a container states as typed ``job.spec`` fields or through its tasks.
_STAGE_TYPED = frozenset(
{"method", "functional", "basis", "charge", "multiplicity", "optimize", "hessian", "aux_basis"}
)
[docs]
def stage_container(
document: Document, recipe_id: str, stage_id: str, species_id: str, *, preview: bool = False
) -> bytes:
"""A pending QVF job container for one stage of a recipe, for one species.
Like :func:`job_container`, with the ``job.spec`` spelled from the
stage's resolved ``run_job`` keywords (:mod:`vibesketch.inputs.recipe`):
the typed fields, the runner's tasks for the stage's kind, and every
other keyword under ``options`` by name. The structure is the species'
first conformer: the campaign's stage driver replaces its coordinates
with the accepted geometry it is staged, so the template never runs on
the sketch's coordinates where a parent stage exists. Raises
:class:`QvfError` for a stage the resolver refuses or a species without
a conformer.
"""
from vibesketch import __version__
from vibesketch.inputs.recipe import resolve_stage
qvf_writer, _ = _modules()
recipe = next((r for r in document.recipes if r.id == recipe_id), None)
if recipe is None:
raise QvfError(f"the document has no recipe {recipe_id!r}")
stage = recipe.stage_by_id(stage_id)
species = document.species_by_id(species_id)
if not species.conformers:
raise QvfError(f"species {species_id} has no 3D structure yet")
resolved = resolve_stage(stage, species)
if resolved.refused:
raise QvfError(
f"stage {stage_id} is refused for {species_id}: " + "; ".join(resolved.refusals)
)
if stage.kind not in _STAGE_TASKS:
raise QvfError(f"stage {stage_id} is a {stage.kind} stage, which runs no calculation")
keywords = dict(resolved.keywords)
fields: dict[str, Any] = {
"job_type": "molecular",
"method": keywords["method"],
"charge": species.charge,
"multiplicity": species.multiplicity,
"tasks": list(_STAGE_TASKS[stage.kind]),
}
if "functional" in keywords:
fields["functional"] = keywords["functional"]
if keywords.get("basis") is not None:
fields["basis"] = keywords["basis"]
if "dispersion" in keywords:
fields["dispersion"] = keywords.get("dispersion")
options = {
key: value
for key, value in keywords.items()
if key not in _STAGE_TYPED and key != "dispersion" and value is not None
}
if keywords.get("aux_basis") is not None:
options["aux_basis"] = keywords["aux_basis"]
if options:
fields["options"] = options
conformer = species.conformers[0]
what = _what(species)
writer = qvf_writer.QvfWriter(
program="vibe-sketch",
version=__version__,
calculation=f"{recipe_id}/{stage_id}: {stage.kind} for {species_id}: {what}",
)
writer.set_extensions({NAMESPACE: {"version": EXTENSION_VERSION, "critical": False}})
members: dict[str, Any] = {
"document": document.to_json_data(),
"species": {"species": _species_index(document)},
"stage": {
"recipe": recipe_id,
"stage": stage_id,
"species": species_id,
"kind": stage.kind,
},
}
writer.add_vendor_section(
DOCUMENT_KIND,
json_members=members,
label="vibe-sketch document",
section_id="vibesketch_document",
)
if preview:
_add_preview(writer, document)
structure = writer.add_structure(
_atoms(conformer),
bonds=_bonds(species, conformer),
section_id=f"structure_{species.id}",
**_identity(species),
)
# Other species are prepared context, never this stage's calculation result.
structures = {species.id}
for other in document.species:
if other.id == species.id or not other.conformers:
continue
writer.add_structure(
_atoms(other.conformers[0]),
bonds=_bonds(other, other.conformers[0]),
section_id=f"structure_{other.id}",
**_identity(other),
)
structures.add(other.id)
steps = _steps(document, structures)
if steps:
writer.add_reaction_scheme(steps, label="reaction scheme (prepared context)")
writer.add_job_spec(
**fields,
structure_ref=structure,
label=f"{recipe_id}/{stage_id}: {stage.kind} for vibeqc",
section_id=f"job_{recipe_id}_{stage_id}",
)
writer.set_provenance(
run_status="pending", charge=species.charge, multiplicity=species.multiplicity
)
return bytes(writer.to_bytes())
# ── Reading ─────────────────────────────────────────────────────────────
def _stated(section: dict[str, Any], provenance: dict[str, Any], single: bool) -> dict[str, int]:
"""The charge and multiplicity a structure section states.
A field the section leaves out states nothing (spec § 5.1); with a single
structure, ``provenance`` supplies it.
"""
stated: dict[str, int] = {}
for key in ("charge", "multiplicity"):
value = section.get(key)
if value is None and single:
value = provenance.get(key)
number = _number(value)
if number is not None:
stated[key] = number
return stated
def _number(value: Any) -> int | None:
return value if isinstance(value, int) and not isinstance(value, bool) else None
def _text(value: Any) -> str | None:
return value if isinstance(value, str) and value else None
def _yield_text(step: dict[str, Any]) -> str | None:
text = step.get("yield_text")
if isinstance(text, str) and text.strip():
return text.strip()
percent = step.get("yield_percent")
if isinstance(percent, int | float) and not isinstance(percent, bool):
return f"{percent:g}%"
return None
def _steps_of(reader: Any, sections: list[dict[str, Any]], warnings: list[str]) -> list[Any]:
"""The steps of a QVF's ``reaction.scheme`` sections (spec § 5.14)."""
from vibesketch.document.coordinates import Step
found = []
for section in sections:
try:
steps = reader.read_json(section["members"]["scheme"])["steps"]
except (KeyError, TypeError, ValueError) as error:
warnings.append(f"{section['id']}: the reaction scheme cannot be read ({error})")
continue
for step in steps:
roles = {
role: [str(p["structure_ref"]) for p in step.get(role) or []]
for role in ("reactants", "products", "reagents")
}
if any(
p.get("coefficient", 1) != 1
for role in ("reactants", "products", "reagents")
for p in step.get(role) or []
):
warnings.append(f"step {step['id']}: its coefficients are not drawn")
found.append(
Step(
id=str(step["id"]),
reactants=roles["reactants"],
products=roles["products"],
reagents=roles["reagents"],
arrow=str(step.get("arrow", "forward")),
conditions=[str(line) for line in step.get("conditions") or []],
yield_text=_yield_text(step),
)
)
return found
def _species_from_structures(reader: Any, sections: list[dict[str, Any]]) -> Derivation:
"""A document drawn from a QVF's ``structure`` and ``reaction.scheme`` sections."""
from vibesketch.document.coordinates import NothingDrawn, Structure, from_structures
structures = [s for s in sections if s.get("kind") == "structure"]
if not structures:
raise QvfError("the QVF holds neither a vibe-sketch document nor a structure")
provenance = reader.manifest.get("provenance") or {}
warnings: list[str] = []
found: list[Structure] = []
for section in structures:
payload = reader.read_json(section["members"]["structure"])
if any(payload.get("pbc") or []):
warnings.append(f"{section['id']} is periodic; vibe-sketch draws molecules only")
continue
stated = _stated(section, provenance, len(structures) == 1)
found.append(
Structure(
id=section["id"],
symbols=[str(a["symbol"]) for a in payload["atoms"]],
positions=[[float(v) for v in a["position"]] for a in payload["atoms"]],
charge=stated.get("charge"),
multiplicity=stated.get("multiplicity"),
label=_text(section.get("label")),
name=_text(section.get("name")),
smiles=_text(section.get("smiles")),
)
)
schemes = [s for s in sections if s.get("kind") == "reaction.scheme"]
steps = _steps_of(reader, schemes, warnings)
try:
return from_structures(found, steps=steps, warnings=warnings, source="qvf")
except NothingDrawn as error:
raise QvfError(str(error)) from None
#: Spellings of the hartree in a ``{value, units}`` object (spec § 4.3 says ``hartree``;
#: vibe-qc writes ``Eh``).
_HARTREE = frozenset({"hartree", "eh", "ha", "e_h", "au", "a.u."})
def _energy(provenance: dict[str, Any]) -> tuple[dict[str, Any] | None, float | None]:
"""``provenance.scf_energy`` as written, and its value in hartree when the units say so."""
energy = provenance.get("scf_energy")
if not isinstance(energy, dict) or not isinstance(energy.get("value"), int | float):
return None, None
value = float(energy["value"])
units = str(energy.get("units", ""))
stated = {"value": value, "units": units}
return stated, value if units.lower() in _HARTREE else None
def _latest_record(sections: list[dict[str, Any]]) -> dict[str, Any] | None:
"""The ``run.record`` of the run that settled the container: the highest sequence."""
records = [s for s in sections if s.get("kind") == "run.record"]
if not records:
return None
return max(enumerate(records), key=lambda r: (_number(r[1].get("sequence")) or 0, r[0]))[1]
def _settled(reader: Any, sections: list[dict[str, Any]], document: Document) -> Derivation:
"""The document a job container holds, with what its runner left behind.
A container is the one ``job.spec`` section with ``provenance.run_status``
stated (spec § 3.2, § 5.9). The runner keeps the input structure and adds
its result structure with a ``run_ref`` to its ``run.record``; that
structure becomes a new conformer of the job's species when its atoms are
the ones submitted, in the same order.
"""
provenance = reader.manifest.get("provenance") or {}
status = _text(provenance.get("run_status")) or "not stated"
(job_section,) = [s for s in sections if s.get("kind") == "job.spec"]
job_id = str(job_section["id"]).removeprefix("job_")
job = next((j for j in document.jobs if j.id == job_id), None)
species_id = job.species if job is not None else None
if species_id is None:
reference = str(job_section.get("structure_ref") or "")
species_id = reference.removeprefix("structure_") or None
warnings: list[str] = []
record = _latest_record(sections)
spec = reader.read_json(job_section["members"]["spec"])
result: dict[str, Any] = {
"job": job_id if job is not None else None,
"species": species_id,
"status": status,
"program": _text(record.get("program")) if record else None,
"program_version": _text(record.get("program_version")) if record else None,
"method": _text(provenance.get("method")) or _text(spec.get("method")),
"tasks": [str(t) for t in spec.get("tasks") or ["single_point"]],
"conformer": None,
"exit_status": _number(record.get("exit_status")) if record else None,
"started_utc": _text(record.get("started_utc")) if record else None,
"finished_utc": _text(record.get("finished_utc")) if record else None,
"wall_seconds": provenance.get("wall_seconds")
if isinstance(provenance.get("wall_seconds"), int | float)
else None,
}
result["energy"], result["energy_hartree"] = _energy(provenance)
label = f"job {job_id}" if job is not None else "the job"
if status == "failed":
warnings.append(f"{label}: the run failed; its log is in the container's run record")
elif status != "converged":
warnings.append(f"{label}: the calculation has not finished (run status {status})")
results = [s for s in sections if s.get("kind") == "structure" and s.get("run_ref")]
if record is not None:
results = [s for s in results if s["run_ref"] == record["id"]] or results
species = next((s for s in document.species if s.id == species_id), None)
if status == "converged" and results and species is not None and species.conformers:
payload = reader.read_json(results[-1]["members"]["structure"])
symbols = [str(a["symbol"]) for a in payload["atoms"]]
if symbols != species.conformers[0].symbols:
warnings.append(
f"{label}: the result's atoms are not the ones submitted for {species_id}, "
"so its geometry is left out"
)
else:
conformer = Conformer(
method="vibeqc:" + "+".join(result["tasks"]),
symbols=symbols,
xyz_angstrom=[[float(v) for v in a["position"]] for a in payload["atoms"]],
)
conformers = [*species.conformers, conformer]
updated = species.model_copy(update={"conformers": conformers})
rebuilt = document.model_copy(
update={"species": [updated if s.id == species.id else s for s in document.species]}
)
document = Document.model_validate(rebuilt.to_json_data())
result["conformer"] = len(conformers) - 1
elif status == "converged" and species is None:
warnings.append(f"{label}: the document has no species {species_id!r} to give the result")
return Derivation(document, warnings, result)
[docs]
def from_qvf(data: bytes) -> Derivation:
"""The document a QVF archive holds; raises :class:`QvfError`.
A job container gives back its result too (``Derivation.result``, the
module docstring says what it holds).
"""
import zipfile
from vibesketch.document.io import DocumentError, from_data
from vibesketch.limits import LimitExceeded, check_archive
try:
check_archive(data)
except LimitExceeded as error:
raise QvfError(str(error)) from None
_, qvf_reader = _modules()
try:
reader = qvf_reader.QvfReader(io.BytesIO(data))
except (zipfile.BadZipFile, KeyError, ValueError) as error:
raise QvfError(f"not a QVF archive ({error})") from None
with reader:
version = reader.manifest.get("qvf_version")
if version not in (1, 2): # the spec asks consumers to take 2 as 1
raise QvfError(f"QVF version {version!r} is newer than this reader")
sections = reader.sections
stored = next((s for s in sections if s.get("kind") == DOCUMENT_KIND), None)
if stored is None:
return _species_from_structures(reader, sections)
try:
payload = reader.read_json(stored["members"]["document"])
except (KeyError, ValueError) as error:
raise QvfError(f"the vibe-sketch document in the QVF is damaged ({error})") from None
try:
document = from_data(payload)
except DocumentError as error:
raise QvfError("; ".join(error.errors)) from None
provenance = reader.manifest.get("provenance") or {}
jobs = [s for s in sections if s.get("kind") == "job.spec"]
if len(jobs) != 1 or provenance.get("run_status") is None:
return Derivation(document) # a document archive: its job set describes, nothing ran
try:
return _settled(reader, sections, document)
except (KeyError, TypeError, ValueError) as error:
raise QvfError(f"the job container's result cannot be read ({error})") from None