Source code for vibesketch.document.model

"""The vibe-sketch document, schema version 1.

A ``.vsk`` file is one JSON object. It carries the canvas state verbatim, in
the engine's own format (KET), so a drawing survives a round trip without
loss, and a normalized layer the compute side reads: species with their
identifiers, charge, multiplicity and 3D conformers; reactions with roles,
arrows and conditions; program-neutral job specifications; the figure style;
and provenance. The design is roadmap section 3.1; ``docs/document.md``
describes every field.

The model is strict. Unknown fields are refused rather than ignored, and the
cross-references (a reaction's species, a job's species) must resolve, so an
agent that writes a document gets told what is wrong instead of having its
data silently dropped. ``schema/vibe-sketch-1.schema.json`` is generated from
these classes (``vibe-sketch schema``) and must not be edited by hand.
"""

from __future__ import annotations

import math
from datetime import UTC, datetime
from typing import Annotated, Any, Final, Literal, Self

from pydantic import (
    AwareDatetime,
    BaseModel,
    ConfigDict,
    Field,
    JsonValue,
    StringConstraints,
    field_validator,
    model_validator,
)

from vibesketch.elements import ATOMIC_NUMBER

__all__ = [
    "MAX_HELM",
    "SCHEMA_ID",
    "Arrow",
    "Canvas",
    "Capping",
    "SolidFormula",
    "Coefficient",
    "Electrons",
    "Quantity",
    "ReferenceElectrode",
    "State",
    "StateLabel",
    "Chirality",
    "Conformer",
    "Defect",
    "Fullerene",
    "Graphene",
    "NanoGroup",
    "NanoSite",
    "Nanoribbon",
    "Nanostructure",
    "NanostructureParameters",
    "Nanotube",
    "PeriodicCell",
    "SidewallGroup",
    "TubeEnd",
    "Generic",
    "Constraint",
    "Document",
    "Edit",
    "Job",
    "JobSpec",
    "Name",
    "Provenance",
    "RGroup",
    "Reaction",
    "Resources",
    "Run",
    "Sequence",
    "Solvent",
    "Species",
    "Style",
    "Surface",
    "SurfaceAnchor",
    "Task",
]

#: The value of a version-1 document's ``schema`` field.
SCHEMA_ID: Final = "vibe-sketch/1"

#: A document-wide identifier: a letter, then letters, digits, ``_`` or ``-``.
_Identifier = Annotated[str, StringConstraints(pattern=r"^[A-Za-z][A-Za-z0-9_-]{0,63}$")]

#: A non-empty line of free text.
_Text = Annotated[str, StringConstraints(min_length=1, max_length=500)]

#: A lower-case token naming a method, program, preset or source.
_Token = Annotated[str, StringConstraints(pattern=r"^[a-z0-9][a-z0-9_.:+-]{0,63}$")]

#: A method name as programs spell it: B3LYP, wB97X-D3, CCSD(T), MP2.
_MethodName = Annotated[str, StringConstraints(pattern=r"^[A-Za-z0-9][A-Za-z0-9()+*/,_.-]{0,63}$")]

#: A basis set name: def2-TZVP, 6-31G*, aug-cc-pVTZ, def2/J.
_BasisName = Annotated[str, StringConstraints(pattern=r"^[A-Za-z0-9][A-Za-z0-9()+*/_.-]{0,63}$")]

#: A short keyword: RIJCOSX, D4, D3(BJ).
_Keyword = Annotated[str, StringConstraints(pattern=r"^[A-Za-z0-9][A-Za-z0-9()_-]{0,31}$")]

#: A solvent as the solvation model names it: water, dichloromethane, n-hexane.
_SolventName = Annotated[str, StringConstraints(pattern=r"^[A-Za-z0-9][A-Za-z0-9 ,()'-]{0,63}$")]

#: A SMILES string: no whitespace.
_Smiles = Annotated[str, StringConstraints(pattern=r"^\S+$")]
#: A file or directory path as the queue reports it.
_Path = Annotated[str, StringConstraints(min_length=1, max_length=4096)]

#: A standard InChIKey.
_InChIKey = Annotated[str, StringConstraints(pattern=r"^[A-Z]{14}-[A-Z]{10}-[A-Z]$")]

#: The arrows a reaction can carry (roadmap WP 3.1).
Arrow = Literal["forward", "equilibrium", "retrosynthetic", "resonance"]

#: What a job computes. ``geometry`` is no calculation: the structure only.
Task = Literal["geometry", "sp", "opt", "freq", "opt+freq", "tddft", "nmr"]


class _Strict(BaseModel):
    model_config = ConfigDict(extra="forbid", populate_by_name=True, validate_assignment=True)


[docs] class Canvas(_Strict): """The drawing as the canvas engine stores it, embedded verbatim.""" format: Literal["ket"] = Field(description="The engine format; KET is Ketcher's JSON.") data: dict[str, JsonValue] = Field( default_factory=dict, description="The KET document, unchanged. Empty when nothing is drawn.", ) @field_validator("data") @classmethod def _ket_has_a_root(cls, data: dict[str, JsonValue]) -> dict[str, JsonValue]: if data and not isinstance(data.get("root"), dict): raise ValueError("a KET document needs a 'root' object") return data
[docs] class Name(_Strict): """A name for a species, with where it came from and how sure it is.""" value: _Text source: _Token = Field( description="Where the name came from, for example 'vibeqc_naming' or 'user'." ) confidence: Literal["high", "medium", "low"] = Field( description="The naming engine's confidence; 'high' for a name the user typed." )
#: The longest HELM a species records, or CXSMILES a generic structure has: #: well beyond the longest sequence vibe-sketch lays out. MAX_HELM = 50_000 #: A CXSMILES: a SMILES, then optionally a space and its |...| extensions. _CXSmiles = Annotated[str, StringConstraints(pattern=r"^\S+( \|[^\n]*\|)?$", max_length=MAX_HELM)]
[docs] class Generic(_Strict): """A structure in the drawing that is not one species (#90). An R-group core, whose R-group atoms are labelled ``_R1`` and so on, a group in polymer brackets (SRU, COP, MUL or GEN), or a structure bonded to labelled attachment atoms that stand for a material, such as ``CNT`` or ``Au`` (#139), as CXSMILES without coordinates. Derived from the drawing like species; the drawing is the source of truth. """ id: _Identifier kind: Literal["rgroup-core", "bracketed", "labelled"] cxsmiles: _CXSmiles label: _Text | None = Field( default=None, description="The compound number in the scheme, for example '1' or '3a'." ) rgroups: list[Annotated[int, Field(ge=1)]] = Field( default_factory=list, description="The R-group numbers a core uses, R1 as 1." ) labels: list[_Text] = Field( default_factory=list, description="A labelled structure's attachment labels, such as 'CNT', in atom order.", )
[docs] class RGroup(_Strict): """An R-group's definition: its substituents and conditions (#90).""" number: int = Field(ge=1, description="1 for R1.") substituents: list[_CXSmiles] = Field( min_length=1, description="Each substituent as CXSMILES, its attachment points labelled _AP1, _AP2.", ) occurrence: _Text | None = Field( default=None, description="How often it occurs, as drawn: '1-3', '>2' or '1,3-5'." ) rest_h: bool = Field(default=False, description="Unsubstituted positions must be hydrogen.") if_then: int | None = Field( default=None, ge=1, description="The R-group number this one requires, if any." )
#: The longest HELM a species records: well beyond the longest sequence vibe-sketch lays out.
[docs] class Sequence(_Strict): """The monomer form of a peptide or nucleic acid the species was built from (#98). It is trusted only for the structure it was built as, which ``inchikey`` names: a species whose drawing has changed loses it. """ helm: Annotated[str, StringConstraints(min_length=1, max_length=MAX_HELM)] = Field( description="The HELM of the peptide or nucleic acid: RDKit's, or as given." ) inchikey: _InChIKey = Field(description="The InChIKey of the structure the HELM builds.")
class ConformerSource(_Strict): """Application provenance of an imported geometry; relaxation is opaque JSON.""" program: str version: str | None = None relaxation: dict[str, Any] | None = None
[docs] class Conformer(_Strict): """One 3D structure of a species: element symbols and Cartesian coordinates.""" method: _Token = Field( description="How the coordinates were made, for example 'etkdg+mmff94' or 'imported'." ) symbols: list[str] = Field(min_length=1, description="Element symbols, one per atom.") xyz_angstrom: list[list[float]] = Field( description="Cartesian coordinates in angstrom, one [x, y, z] per atom." ) source: ConformerSource | None = None @field_validator("symbols") @classmethod def _known_elements(cls, symbols: list[str]) -> list[str]: unknown = sorted({s for s in symbols if s not in ATOMIC_NUMBER}) if unknown: raise ValueError(f"unknown element symbols: {', '.join(unknown)}") return symbols @field_validator("xyz_angstrom") @classmethod def _finite_triples(cls, rows: list[list[float]]) -> list[list[float]]: for index, row in enumerate(rows): if len(row) != 3: raise ValueError(f"atom {index} has {len(row)} coordinates, not 3") if not all(math.isfinite(value) for value in row): raise ValueError(f"atom {index} has a non-finite coordinate") return rows @model_validator(mode="after") def _one_position_per_atom(self) -> Self: if len(self.xyz_angstrom) != len(self.symbols): raise ValueError(f"{len(self.symbols)} symbols but {len(self.xyz_angstrom)} positions") return self
[docs] def electron_count(self, charge: int) -> int: """Electrons in the neutral atoms, less the charge.""" return sum(ATOMIC_NUMBER[s] for s in self.symbols) - charge
[docs] class SolidFormula(_Strict): """A solid given by its formula, without a molecular graph (#142). Oxides, salts, catalysts and scales: Fe2O3, Pb3(CO3)2(OH)2, CuSO4·5H2O, Fe3−xCrxO4, γ-Al2O3. ``key`` (the canonical formula with polymorph and state) identifies it instead of an InChIKey. Nothing is calculated for it: that needs its structure (roadmap Phase 5). """ text: _Text = Field(description="The formula as drawn, such as 'γ-Al2O3(s)'.") formula: Annotated[str, StringConstraints(min_length=1, max_length=200)] = Field( description="Each phase in Hill order, a variable count in parentheses: Cr(x)Fe(3-x)O4." ) key: Annotated[str, StringConstraints(min_length=1, max_length=220)] polymorph: Annotated[str, StringConstraints(min_length=1, max_length=1)] | None = None state: Literal["s", "l", "g", "aq", "ads"] | None = None composition: dict[str, float | str] = Field( description="Atoms per formula unit, all phases together; a variable count as written." ) molar_mass_g_mol: float | None = Field( default=None, gt=0, allow_inf_nan=False, description="Absent with a variable count." ) @model_validator(mode="after") def _reads_as_written(self) -> Self: from vibesketch.chem.formula import FormulaError, parse_formula try: read = parse_formula(self.text) except FormulaError as error: raise ValueError(str(error)) from None if (read.canonical(), read.key()) != (self.formula, self.key): raise ValueError(f"the solid {self.text!r} reads as {read.key()}, not {self.key}") return self
[docs] class Capping(_Strict): """How a labelled generic structure became a calculable species: its cap (#139). Each attachment atom (CNT, Au, ...) of the generic structure ``cxsmiles`` was replaced by ``cap``, as the user chose; nothing is capped unasked. """ generic: _Identifier = Field(description="The labelled generic structure it was made from.") cxsmiles: _CXSmiles = Field(description="That structure, so a redraw can find it again.") cap: Literal["H", "CH3"] labels: list[_Text] = Field(min_length=1, description="The labels the cap replaced.")
#: A group bonded to a nanostructure's carbon: an edge or end termination or a sidewall group. NanoGroup = Literal["H", "OH", "COOH"] #: How a nanotube end is finished: terminated by a group, or closed with a fullerene hemisphere. TubeEnd = Literal["H", "OH", "COOH", "capped"]
[docs] class Chirality(_Strict): """A nanotube wall's chiral indices (n, m), with 0 <= m <= n.""" n: int = Field(ge=1, le=100) m: int = Field(default=0, ge=0, le=100) @model_validator(mode="after") def _ordered(self) -> Self: if self.m > self.n: raise ValueError(f"({self.n},{self.m}): write the indices with m <= n") return self
[docs] class NanoSite(_Strict): """Where on a nanostructure a group or defect goes; the nearest carbon is taken. ``along`` is the fraction of the length; ``around`` is the angle about a nanotube's axis in degrees, ``across`` the fraction of a sheet's or ribbon's width. """ along: float = Field(ge=0, le=1, allow_inf_nan=False) around: float | None = Field(default=None, ge=0, lt=360, allow_inf_nan=False) across: float | None = Field(default=None, ge=0, le=1, allow_inf_nan=False) @model_validator(mode="after") def _one_crosswise(self) -> Self: if (self.around is None) == (self.across is None): raise ValueError("a site gives 'around' (a nanotube) or 'across' (a sheet), not both") return self
[docs] class SidewallGroup(_Strict): """A group bonded to a basal-plane carbon, which becomes sp3.""" group: NanoGroup site: NanoSite
[docs] class Defect(_Strict): """A lattice defect: a missing carbon, or a C-C bond turned by 90 degrees (Stone-Wales).""" kind: Literal["vacancy", "stone-wales"] site: NanoSite
[docs] class Nanotube(_Strict): """A single- or multi-walled carbon nanotube segment.""" kind: Literal["nanotube"] walls: list[Chirality] = Field( min_length=1, max_length=10, description="Chiral indices, innermost wall first." ) length_nm: float | None = Field( default=None, gt=0, le=1000, allow_inf_nan=False, description="Length; or give periods." ) periods: int | None = Field( default=None, ge=1, le=10_000, description="Translation periods of the innermost wall." ) ends: tuple[TubeEnd, TubeEnd] = Field( default=("H", "H"), description="How the two ends are finished, lower end first." ) sidewall: list[SidewallGroup] = Field(default_factory=list, max_length=200) defects: list[Defect] = Field(default_factory=list, max_length=200) @model_validator(mode="after") def _one_length(self) -> Self: if (self.length_nm is None) == (self.periods is None): raise ValueError("give a nanotube's length either in nm or in translation periods") return self
[docs] class Graphene(_Strict): """A rectangular graphene flake: zigzag edges along its width, armchair along its length.""" kind: Literal["graphene"] width_nm: float = Field(gt=0, le=100, allow_inf_nan=False) length_nm: float = Field(gt=0, le=100, allow_inf_nan=False) edges: NanoGroup = "H" sidewall: list[SidewallGroup] = Field(default_factory=list, max_length=200) defects: list[Defect] = Field(default_factory=list, max_length=200)
[docs] class Nanoribbon(_Strict): """A graphene nanoribbon, N-AGNR or N-ZGNR, a whole number of periods long.""" kind: Literal["nanoribbon"] edge: Literal["armchair", "zigzag"] width: int = Field( ge=2, le=100, description="Dimer lines (armchair) or zigzag chains across the ribbon." ) periods: int = Field(ge=1, le=10_000) edges: NanoGroup = "H" sidewall: list[SidewallGroup] = Field(default_factory=list, max_length=200) defects: list[Defect] = Field(default_factory=list, max_length=200)
[docs] class Fullerene(_Strict): """Buckminsterfullerene C60 (Ih) or C70 (D5h).""" kind: Literal["fullerene"] size: Literal[60, 70]
NanostructureParameters = Annotated[ Nanotube | Graphene | Nanoribbon | Fullerene, Field(discriminator="kind") ]
[docs] class PeriodicCell(_Strict): """The one-dimensional cell a finite segment repeats, recorded for periodic models. Nothing writes it yet; it is what a periodic calculation of the infinite structure needs (roadmap Phase 5). """ axis: Literal["z"] = "z" translation_angstrom: float = Field(gt=0, allow_inf_nan=False) carbons_per_period: int = Field(ge=1)
[docs] class Nanostructure(_Strict): """How a generated species was made: the parameters, and the graph its coordinates lack. The atoms are the species' first conformer (method ``nanostructure``); ``bonds`` connect them. A nanostructure has no SMILES or InChI; ``key`` identifies it instead, from the parameters alone. """ parameters: NanostructureParameters identity: Annotated[str, StringConstraints(min_length=1, max_length=4000)] = Field( description="The canonical form of the parameters, which the key digests." ) key: Annotated[str, StringConstraints(pattern=r"^NANO-[0-9A-F]{20}$")] description: _Text formula: Annotated[str, StringConstraints(pattern=r"^([A-Z][a-z]?[0-9]*)+$")] bonds: list[tuple[Annotated[int, Field(ge=0)], Annotated[int, Field(ge=0)], float]] = Field( description="[i, j, order] over the first conformer's atoms; order 1, 1.5 or 2." ) periodic: PeriodicCell | None = None @model_validator(mode="after") def _key_matches(self) -> Self: from vibesketch.chem.nanostructures import identity_of identity, key = identity_of(self.parameters) if (identity, key) != (self.identity, self.key): raise ValueError( f"the nanostructure's identity {self.key} does not match its parameters ({key})" ) return self
[docs] class Species(_Strict): """A molecule or ion that appears in the drawing.""" id: _Identifier label: _Text | None = Field( default=None, description="The compound number in the scheme, for example '1' or '3a'." ) smiles: _Smiles | None = None inchikey: _InChIKey | None = None charge: int = Field(default=0, description="Net charge: the sum of the formal charges.") multiplicity: int = Field( default=1, ge=1, description="Spin multiplicity 2S+1. Defaults from the radical count; always editable.", ) name: Name | None = None conformers: list[Conformer] = Field(default_factory=list) sequence: Sequence | None = Field( default=None, description="The sequence it was built from; absent when drawn by hand." ) nanostructure: Nanostructure | None = Field( default=None, description="The generator parameters and bonds of a generated nanostructure.", ) capped: Capping | None = Field( default=None, description="The labelled structure and cap it was made from, when capped on request.", ) solid: SolidFormula | None = Field( default=None, description="A solid given by its formula, without a molecular graph." ) @model_validator(mode="after") def _sequence_is_this_structure(self) -> Self: if self.sequence is not None and self.sequence.inchikey != self.inchikey: raise ValueError( f"the sequence was recorded for {self.sequence.inchikey}, " f"not this species' structure ({self.inchikey})" ) return self @model_validator(mode="after") def _a_solid_has_no_graph(self) -> Self: if self.solid is not None and ( self.smiles is not None or self.inchikey is not None or self.conformers or self.nanostructure is not None ): raise ValueError( f"species {self.id} is the solid {self.solid.key}: it has no SMILES, InChIKey, " "conformer or nanostructure" ) return self @model_validator(mode="after") def _nanostructure_is_its_first_conformer(self) -> Self: made = self.nanostructure if made is None: return self if self.smiles is not None or self.inchikey is not None: raise ValueError("a nanostructure is identified by its key, not a SMILES or InChIKey") if not self.conformers: raise ValueError("a nanostructure species needs its atoms as conformer 0") symbols = self.conformers[0].symbols from vibesketch.chem.nanostructures import hill_formula if hill_formula(symbols) != made.formula: raise ValueError( f"conformer 0 is {hill_formula(symbols)}, not the nanostructure's {made.formula}" ) for i, j, order in made.bonds: if not i < j < len(symbols) or order not in (1.0, 1.5, 2.0): raise ValueError(f"the nanostructure bond [{i}, {j}, {order}] is not valid") if len({(i, j) for i, j, _ in made.bonds}) != len(made.bonds): raise ValueError("a nanostructure bond is listed twice") return self @model_validator(mode="after") def _conformers_agree_with_charge_and_spin(self) -> Self: if not self.conformers: return self first = self.conformers[0].symbols for index, conformer in enumerate(self.conformers[1:], start=1): if conformer.symbols != first: raise ValueError(f"conformer {index} has different atoms from conformer 0") electrons = self.conformers[0].electron_count(self.charge) unpaired = self.multiplicity - 1 if electrons < 0: raise ValueError(f"charge {self.charge} leaves {electrons} electrons") if unpaired > electrons or (electrons - unpaired) % 2: raise ValueError( f"multiplicity {self.multiplicity} is impossible with {electrons} electrons " f"(charge {self.charge}): the number of unpaired electrons must have the " "parity of the electron count" ) return self
_Positive = Annotated[float, Field(gt=0, allow_inf_nan=False)] class Amount(_Strict): """What was entered for one participant of a reaction; nothing is inferred from the drawing. At most one of ``mmol``, ``mass_mg`` and ``volume_ml`` gives the amount; a volume needs a density or a concentration. ``equivalents`` alone scales the reference amount. Density and concentration may also be given without a volume, so that the volume to measure out is worked out. """ species: _Identifier coefficient: _Positive = Field(default=1.0, description="Stoichiometric coefficient.") equivalents: _Positive | None = None mmol: _Positive | None = None mass_mg: _Positive | None = None volume_ml: _Positive | None = None density_g_ml: _Positive | None = None concentration_mol_l: _Positive | None = None @model_validator(mode="after") def _one_way_to_the_amount(self) -> Self: given = [n for n in ("mmol", "mass_mg", "volume_ml") if getattr(self, n) is not None] if len(given) > 1: raise ValueError(f"{self.species}: give one of mmol, mass_mg or volume_ml, not both") if self.density_g_ml is not None and self.concentration_mol_l is not None: raise ValueError(f"{self.species}: give a density or a concentration, not both") if self.volume_ml is not None and ( self.density_g_ml is None and self.concentration_mol_l is None ): raise ValueError(f"{self.species}: a volume needs a density or a concentration") if given and self.equivalents is not None: raise ValueError( f"{self.species}: give an amount or equivalents; the other is worked out" ) return self
[docs] class Coefficient(_Strict): """A participant's stoichiometric coefficient as drawn, such as the 2 of 2H2O (#141).""" species: _Identifier value: _Positive
#: The states a participant's label can give: IUPAC's (s), (l), (g), (aq), and (ads) #: for an adsorbed species, also written with an asterisk. State = Literal["s", "l", "g", "aq", "ads"]
[docs] class StateLabel(_Strict): """A participant's state as drawn beside it, such as Pb(s) or H+(aq) (#141).""" species: _Identifier state: State
[docs] class Electrons(_Strict): """The electrons a half-reaction takes up or gives off, as drawn: 2e⁻ (#141). Electrons are no species, as SMILES cannot write one; they are counted in the charge balance on their ``side``. """ side: Literal["reactants", "products"] count: int = Field(ge=1, le=100)
#: The reference electrodes a potential can be stated against. ReferenceElectrode = Literal["SHE", "NHE", "SCE", "RHE", "Ag/AgCl"]
[docs] class Quantity(_Strict): """A condition at the arrow read with its unit: potential, pH, temperature, time (#141).""" kind: Literal["potential", "ph", "temperature", "time"] value: float = Field(allow_inf_nan=False) unit: Literal["V", "pH", "°C", "K", "s", "min", "h", "d"] reference: ReferenceElectrode | None = Field( default=None, description="A potential's reference electrode." ) kcl_mol_l: _Positive | None = Field( default=None, description="The KCl concentration of an Ag/AgCl reference." ) text: _Text = Field(description="The condition line it was read from.") @model_validator(mode="after") def _reference_with_potential(self) -> Self: if (self.kind == "potential") != (self.reference is not None): raise ValueError("a potential, and only a potential, names its reference electrode") if self.kcl_mol_l is not None and self.reference != "Ag/AgCl": raise ValueError("only an Ag/AgCl reference has a KCl concentration") return self
[docs] class Reaction(_Strict): """One step of a scheme: who reacts, under which conditions, to give what.""" id: _Identifier reactants: list[_Identifier] = Field(min_length=1) reagents: list[_Identifier] = Field( default_factory=list, description="Species drawn above or below the arrow." ) products: list[_Identifier] = Field(min_length=1) arrow: Arrow = "forward" conditions: list[_Text] = Field( default_factory=list, description="Text lines at the arrow, in reading order." ) yield_: _Text | None = Field(default=None, alias="yield", description="As printed, e.g. '87%'.") amounts: list[Amount] = Field( default_factory=list, description="What the chemist entered for the stoichiometry table, per participant.", ) coefficients: list[Coefficient] = Field( default_factory=list, description="Coefficients drawn before participants; 1 otherwise." ) electrons: Electrons | None = Field( default=None, description="Electrons drawn as a participant of a half-reaction." ) states: list[StateLabel] = Field( default_factory=list, description="State labels drawn beside participants." ) quantities: list[Quantity] = Field( default_factory=list, description="The conditions that read as a value with its unit." ) @model_validator(mode="after") def _no_species_twice_in_a_role(self) -> Self: for role in ("reactants", "reagents", "products"): members: list[str] = getattr(self, role) if len(members) != len(set(members)): raise ValueError(f"a species appears twice among the {role}") return self @model_validator(mode="after") def _amounts_of_participants(self) -> Self: named = [amount.species for amount in self.amounts] if len(named) != len(set(named)): raise ValueError("a species has more than one amount") strangers = sorted(set(named) - set(self.species_ids())) if strangers: raise ValueError(f"amounts for species not in the reaction: {', '.join(strangers)}") return self @model_validator(mode="after") def _labels_of_participants(self) -> Self: present = set(self.species_ids()) for what, entries in (("coefficients", self.coefficients), ("states", self.states)): named = [entry.species for entry in entries] if len(named) != len(set(named)): raise ValueError(f"a species has more than one entry in {what}") strangers = sorted(set(named) - present) if strangers: raise ValueError(f"{what} for species not in the reaction: {', '.join(strangers)}") return self
[docs] def species_ids(self) -> list[str]: return [*self.reactants, *self.reagents, *self.products]
[docs] def coefficient(self, species: str) -> float: """The drawn coefficient of ``species``, 1 when none is drawn.""" return next((c.value for c in self.coefficients if c.species == species), 1.0)
[docs] class Solvent(_Strict): """An implicit solvation model and the solvent it describes.""" model: Literal["CPCM", "SMD"] name: _SolventName
[docs] class Resources(_Strict): """What a job may use on the machine that runs it.""" nprocs: int = Field(default=1, ge=1) maxcore_mb: int | None = Field(default=None, ge=1, description="Memory per core in MB.")
#: Atoms each kind of constraint names. _CONSTRAINT_ATOMS: Final = {"distance": 2, "angle": 3, "dihedral": 4, "position": 1}
[docs] class Constraint(_Strict): """A coordinate an optimisation holds fixed. Atoms are 0-based indices into the species' conformer: the drawing's atoms first, then the hydrogens. """ kind: Literal["distance", "angle", "dihedral", "position"] atoms: list[Annotated[int, Field(ge=0)]] = Field( description="2 atoms for a distance, 3 for an angle, 4 for a dihedral, 1 for a position." ) value: float | None = Field( default=None, description="The target in angstrom or degrees; the starting geometry's value if left out.", ) @model_validator(mode="after") def _atoms_fit_the_kind(self) -> Self: wanted = _CONSTRAINT_ATOMS[self.kind] if len(self.atoms) != wanted: raise ValueError( f"a {self.kind} constraint names {wanted} atoms, not {len(self.atoms)}" ) if len(set(self.atoms)) != len(self.atoms): raise ValueError("a constraint names an atom twice") if self.kind == "position" and self.value is not None: raise ValueError("a position constraint fixes an atom where it is; it takes no value") if self.value is not None and not math.isfinite(self.value): raise ValueError("a constraint value must be finite") if self.kind == "distance" and self.value is not None and self.value <= 0: raise ValueError("a distance must be positive") return self
[docs] class JobSpec(_Strict): """A program-neutral description of a calculation. Writers translate it into one program's input and refuse, with a reason, whatever they cannot express (AGENTS.md, ground rule 5). The model only checks that the specification is consistent in itself. """ task: Task method: _MethodName | None = None basis: _BasisName | None = None aux_basis: _BasisName | None = None approximation: _Keyword | None = None dispersion: _Keyword | None = None solvent: Solvent | None = None states: int | None = Field( default=None, ge=1, le=200, description="Excited states to compute; TDDFT only." ) constraints: list[Constraint] | None = Field( default=None, description="Coordinates held fixed; optimisations only." ) resources: Resources | None = None @model_validator(mode="after") def _a_calculation_names_its_method(self) -> Self: if self.task == "geometry": carried = [n for n in self.calculation_fields() if getattr(self, n) is not None] if carried: raise ValueError( f"task 'geometry' is no calculation, but sets {', '.join(carried)}" ) elif self.method is None: raise ValueError(f"task {self.task!r} needs a method") if self.task == "tddft" and self.states is None: raise ValueError("task 'tddft' needs the number of excited states (states)") if self.states is not None and self.task != "tddft": raise ValueError(f"states only means something for task 'tddft', not {self.task!r}") if self.constraints and self.task not in ("opt", "opt+freq"): raise ValueError( f"constraints only mean something for an optimisation, not {self.task!r}" ) return self
[docs] @staticmethod def calculation_fields() -> tuple[str, ...]: """The fields that only mean something for a calculation.""" return ( "method", "basis", "aux_basis", "approximation", "dispersion", "solvent", "states", "constraints", )
class ResultEnergy(_Strict): """An absolute energy, with its location in the source output.""" value: float = Field(allow_inf_nan=False) units: Literal["hartree"] = "hartree" source: _Text class ResultThermochemistry(_Strict): """Absolute thermal energies at one stated temperature and pressure.""" temperature_k: float = Field(gt=0, allow_inf_nan=False) pressure_atm: float = Field(gt=0, allow_inf_nan=False) enthalpy: ResultEnergy gibbs: ResultEnergy class ResultMode(_Strict): """One harmonic vibration; an intensity the output does not give is null, never zero.""" mode: int = Field(ge=0, description="The program's index of the mode.") wavenumber: float = Field( allow_inf_nan=False, description="Harmonic wavenumber; negative for an imaginary mode." ) ir_intensity: float | None = Field(default=None, ge=0, allow_inf_nan=False) raman_activity: float | None = Field(default=None, ge=0, allow_inf_nan=False) class ResultVibrations(_Strict): """The harmonic vibrations a frequency result lists, with their IR and Raman lines. Projected translations and rotations, which the program prints as zero, are left out. """ source: _Text wavenumber_units: Literal["cm-1"] = "cm-1" ir_units: Literal["km/mol"] = "km/mol" raman_units: Literal["angstrom^4/amu"] = "angstrom^4/amu" modes: list[ResultMode] = Field(min_length=1) @property def imaginary(self) -> int: """How many modes are imaginary: none at a minimum, one at a transition state.""" return sum(1 for mode in self.modes if mode.wavenumber < 0) class ResultNucleus(_Strict): """One nucleus's isotropic shielding, by its atom index in the job's input.""" atom: int = Field(ge=0, description="The atom's index in the input, from 0.") element: str isotropic: float = Field(allow_inf_nan=False) anisotropy: float | None = Field(default=None, allow_inf_nan=False) class ResultShieldings(_Strict): """Absolute NMR shieldings of every nucleus, as the program printed them.""" source: _Text units: Literal["ppm"] = "ppm" nuclei: list[ResultNucleus] = Field(min_length=1) class ResultState(_Strict): """One excited state; an oscillator strength the output does not give is null.""" state: _Text = Field(description="The state as the source names it, for example '1-1A'.") energy: float = Field(gt=0, allow_inf_nan=False, description="Excitation energy.") oscillator_strength: float | None = Field(default=None, ge=0, allow_inf_nan=False) class ResultExcitedStates(_Strict): """The excited states a result lists, with their absorption lines.""" source: _Text method: _Text = Field(description="The excited-state method as the source names it.") energy_units: Literal["eV"] = "eV" states: list[ResultState] = Field(min_length=1) class ResultFile(_Strict): """A content-addressed result artifact, never a client-controlled server path.""" name: Annotated[ str, StringConstraints(min_length=1, max_length=200, pattern=r"^[^/\\\x00-\x1f\x7f]+$") ] sha256: Annotated[str, StringConstraints(pattern=r"^[0-9a-f]{64}$")] class RunResult(_Strict): """Verified output bound to the job and input species, not a reaction delta.""" retrieved: AwareDatetime parser: Literal["orca-text/1", "qvf/1"] program_version: _Text source_name: Annotated[ str, StringConstraints(min_length=1, max_length=200, pattern=r"^[^/\\\x00-\x1f\x7f]+$"), ] source_sha256: Annotated[str, StringConstraints(pattern=r"^[0-9a-f]{64}$")] species_sha256: Annotated[str, StringConstraints(pattern=r"^[0-9a-f]{64}$")] specification: JobSpec energy: ResultEnergy view_files: list[ResultFile] = Field(default_factory=list) thermochemistry: ResultThermochemistry | None = None vibrations: ResultVibrations | None = None excited_states: ResultExcitedStates | None = None shieldings: ResultShieldings | None = None
[docs] class Run(_Strict): """One submission of a job to a queue.""" queue: Literal["vibe-queue"] = "vibe-queue" host: Annotated[str, StringConstraints(pattern=r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,63}$")] = ( "localhost" ) id: Annotated[str, StringConstraints(pattern=r"^[A-Za-z0-9_-]{1,64}$")] = Field( description="The queue's id for the run." ) submitted: AwareDatetime directory: _Path | None = Field( default=None, description="The queue's working directory for the run, where the program's output is.", ) container: _Path | None = Field( default=None, description=( "For a vibe-qc QVF job container, its file name in the directory; the finished " "container keeps the sketch and the result." ), ) group: _Identifier | None = Field( default=None, description=( "The group of runs this one was submitted with, such as a reaction's species " "under one specification; the group is followed and cancelled as one." ), ) result: RunResult | None = Field( default=None, description="Verified absolute result with source provenance." )
[docs] class Job(_Strict): """A calculation to run on one species with one program.""" id: _Identifier species: _Identifier program: _Token = Field(description="The writer that produces the input, e.g. 'orca'.") spec: JobSpec runs: list[Run] | None = Field( default=None, description="Submissions to a queue, oldest first." )
#: A SHA-256 digest in lower-case hex. _Sha256 = Annotated[str, StringConstraints(pattern=r"^[0-9a-f]{64}$")] StageKind = Literal["prepare", "mace", "dft", "hessian", "dlpno", "calculation"] class GeometryRef(_Strict): """Where a stage takes its starting geometry. Either the species' own conformer (``conformer``, an index into the species' conformers, 0 for the one the writers use) or the accepted final geometry of an earlier stage (``stage``); exactly one of the two. """ stage: _Identifier | None = Field( default=None, description="The stage whose accepted final geometry this one starts from." ) conformer: int | None = Field( default=None, ge=0, description="The species' conformer to start from, 0 for the first." ) @model_validator(mode="after") def _one_source(self) -> Self: if (self.stage is None) == (self.conformer is None): raise ValueError("a geometry names a stage or a conformer, exactly one of them") return self class MaceOptions(_Strict): """A MACE optimisation: the model, its device and precision, the convergence.""" model: _Token = Field( default="off23-medium", description=( "The MACE model key or alias vibe-qc resolves, such as 'off23-medium' or " "'medium-mpa-0'. A model's licence acknowledgement is not part of a recipe: an " "eligible operator gives it when the recipe is launched." ), ) weight_sha256: _Sha256 | None = Field( default=None, description="The expected SHA-256 of the model weights; a mismatch fails before loading.", ) device: Literal["cpu", "cuda", "mps"] = "cpu" dtype: Literal["float64", "float32"] = Field( default="float64", description="float64 for geometries and energies; MPS needs float32." ) fmax_ev_per_angstrom: float = Field( default=0.02, gt=0, description="Converged when the maximum atomic force vector norm is at or below this.", ) max_steps: int = Field(default=200, ge=1) @model_validator(mode="after") def _mps_is_single_precision(self) -> Self: if self.device == "mps" and self.dtype != "float32": raise ValueError("a MACE run on mps needs dtype float32") return self class DftSurface(_Strict): """A molecular Kohn-Sham surface: what energies, gradients and Hessians are taken on.""" functional: _MethodName = "pbe" basis: _BasisName = "def2-svp" dispersion: _Keyword | None = Field( default=None, description=( "A dispersion correction, D3BJ or D4; none for the publication baseline. The " "vibe-qc route refuses one until its Hessian propagates it (vibe-qc #548)." ), ) density_fit: bool = Field(default=False, description="RI-J; off for the conventional SCF.") cosx: bool = Field(default=False, description="Seminumerical exchange; off for the baseline.") grid: Literal["fine", "orca-defgrid1", "orca-defgrid2", "orca-defgrid3"] = "orca-defgrid3" scf_max_iterations: int | None = Field(default=None, ge=1) stability_max_iterations: int | None = Field(default=None, ge=1) scf_energy_tolerance: float = Field(default=1e-12, gt=0, description="SCF energy change, Eh.") scf_gradient_tolerance: float = Field( default=1e-9, gt=0, description="SCF orbital gradient norm." ) class DftOptimization(_Strict): """A geometry optimisation on a Kohn-Sham surface.""" surface: DftSurface = Field(default_factory=DftSurface) optimizer: Literal["ase"] = Field(default="ase", description="vibe-qc's ASE optimiser route.") fmax_ev_per_angstrom: float = Field(default=0.005, gt=0) max_steps: int = Field(default=300, ge=1) class HessianOptions(_Strict): """A finite-difference Hessian on a Kohn-Sham surface, at the parent's accepted geometry.""" surface: DftSurface = Field(default_factory=DftSurface) displacement_bohr: float = Field( default=0.005, gt=0, description="The central-difference step, in bohr." ) require_positive_modes: bool = False compare_parent_max_delta_cm: float | None = Field(default=None, gt=0) imaginary_mode_threshold_cm: float = Field( default=-20.0, lt=0, description=( "A mode below this wavenumber fails the minimum check; modes between it and zero " "call for a displacement, grid and convergence review, not silent discarding." ), ) class DlpnoOptions(_Strict): """A DLPNO coupled-cluster single point at the parent's accepted geometry.""" method: Literal["dlpno-ccsd(t)", "dlpno-ccsd"] = "dlpno-ccsd(t)" scf_max_iterations: int | None = Field(default=None, ge=1) stability_max_iterations: int | None = Field(default=None, ge=1) scf_energy_tolerance: float | None = Field(default=None, gt=0) scf_gradient_tolerance: float | None = Field(default=None, gt=0) cc_max_iterations: int | None = Field(default=None, ge=1) cc_energy_tolerance: float | None = Field(default=None, gt=0) cc_residual_tolerance: float | None = Field(default=None, gt=0) reference_contract: Literal["all_tested_channels", "real_restricted_cc"] = Field( default="all_tested_channels", description=( "Explicit real restricted CC admission requires verified internal stability and " "resolved real internal/external channels. Generic usability stays separate; " "complex stability and correlated accuracy are not certified." ), ) basis: _BasisName = "cc-pvdz" aux_basis: _BasisName | None = Field( default=None, description="The fitting basis; vibe-qc chooses one for the orbital basis when absent.", ) thresholds: Literal["LoosePNO", "NormalPNO", "TightPNO"] = Field( default="TightPNO", description="The published PNO threshold preset; every resolved threshold is recorded.", ) frozen_core: Literal["chemical", "all-electron"] = Field( default="chemical", description="'chemical': the published per-element core counts; 'all-electron': none.", ) triples_mode: Literal["t0", "t1"] = Field( default="t1", description=( "The (T) correction. In vibe-qc 't1' is a rotated-occupied compatibility " "correction, not an iterative local T1 algorithm; report it as such." ), ) class CalculationStage(_Strict): """One ordinary writer job, including combined optimisation/frequencies.""" program: Literal["orca", "vibeqc"] spec: JobSpec class CitedValue(_Strict): """An explicit thermodynamic choice; units are supplied by its field name.""" value: float = Field(allow_inf_nan=False) source: _Text class RedoxAnalysis(_Strict): """Reduction thermodynamics at 298.15 K, with explicit electron/proton conventions.""" kind: Literal["redox"] reaction: _Identifier absolute_she_v: CitedValue electron_gibbs_kj_mol: CitedValue proton_gas_gibbs_kj_mol: CitedValue | None = None proton_solvation_kj_mol: CitedValue | None = None reference: Literal["SHE", "Ag/AgCl (sat. KCl)", "Ag/AgCl (3 M KCl)", "SCE", "RHE"] ph: float = Field(ge=0, le=14, allow_inf_nan=False) temperature_k: float = Field(ge=298.15, le=298.15, allow_inf_nan=False) input_standard_state: Literal["1 atm"] solution_standard_state: Literal["1 M"] @model_validator(mode="after") def _proton_pair(self) -> Self: if (self.proton_gas_gibbs_kj_mol is None) != (self.proton_solvation_kj_mol is None): raise ValueError("provide both proton gas Gibbs energy and solvation free energy") return self class BindingAnalysis(_Strict): """Finite-cluster binding, with explicit free fragments and complex.""" kind: Literal["binding"] cluster: _Identifier adsorbate: _Identifier complex: _Identifier surface: _Identifier | None = None counterpoise: bool RecipeAnalysis = Annotated[RedoxAnalysis | BindingAnalysis, Field(discriminator="kind")] class Stage(_Strict): """One step of a recipe: what runs, after which stage, from which geometry. ``kind`` names the calculation and exactly one options field, the one of that kind, is set; ``prepare`` is the sketch's own force-field preparation and has none. ``parent`` is the stage whose scientific acceptance this one waits for; ``geometry`` is where its coordinates come from, which is the parent for every numerical stage of the publication baseline. """ id: _Identifier kind: StageKind label: _Text | None = None parent: _Identifier | None = Field( default=None, description="The stage this one follows; none for the first." ) geometry: GeometryRef mace: MaceOptions | None = None dft: DftOptimization | None = None hessian: HessianOptions | None = None dlpno: DlpnoOptions | None = None calculation: CalculationStage | None = None @model_validator(mode="after") def _options_match_the_kind(self) -> Self: given = [ name for name in ("mace", "dft", "hessian", "dlpno", "calculation") if getattr(self, name) ] if self.kind == "prepare": if given: raise ValueError(f"stage {self.id}: a prepare stage takes no {given[0]} options") if self.geometry.conformer is None: raise ValueError(f"stage {self.id}: a prepare stage starts from a conformer") elif given != [self.kind]: raise ValueError( f"stage {self.id}: a {self.kind} stage needs its {self.kind} options and no others" ) return self def yields_geometry(self) -> bool: """Whether an accepted run of this stage produces a geometry others can start from.""" return self.kind in ("prepare", "mace", "dft") or ( self.calculation is not None and self.calculation.spec.task in ("opt", "opt+freq") ) class CampaignRecord(_Strict): """A vibe-queue campaign compiled from a recipe: where it is and what its nodes are. The campaign's attempts, retries and verified artifacts live in vibe-queue's journal in the campaign directory; this record is how the document finds them. """ id: _Identifier #: The directory holding ``campaign.json``, the node sources and the run directory. directory: _Path #: vibe-queue's digest of the campaign file, which every attempt's idempotency key #: derives from; absent when vibe-queue was not installed where it was compiled. campaign_sha256: _Sha256 | None = None created: AwareDatetime model_license_acknowledged: bool = Field( default=False, description=( "Whether the operator who compiled the campaign acknowledged a MACE model's " "licence; the acknowledgement is in the compiled stage containers, never the recipe." ), ) #: Per node id: the species, stage and node kind it stands for. nodes: dict[str, dict[str, str]] = Field(default_factory=dict) class Recipe(_Strict): """A saved, reviewable staged workflow for species of the document. The stages form a chain or tree by ``parent``; identifiers are stable so runs, artifacts and retries can name them. A geometry comes from the species' conformer or from an earlier stage that yields one; a Hessian runs on the same surface as the optimisation it follows. """ id: _Identifier name: _Text | None = None species: list[_Identifier] = Field( description="The species the recipe applies to; every stage runs once per species." ) stages: list[Stage] = Field(min_length=1) runtime_version: Literal[">=0.18,<0.19"] | None = None analysis: RecipeAnalysis | None = None campaigns: list[CampaignRecord] = Field( default_factory=list, description="The vibe-queue campaigns compiled from this recipe." ) @model_validator(mode="after") def _stages_form_a_dag(self) -> Self: seen: dict[str, Stage] = {} for stage in self.stages: if stage.id in seen: raise ValueError(f"recipe {self.id}: stage id {stage.id!r} is used twice") if stage.parent is not None and stage.parent not in seen: raise ValueError( f"recipe {self.id}: stage {stage.id} follows {stage.parent!r}, which is " "not an earlier stage" ) source = stage.geometry.stage if source is not None: if source not in seen: raise ValueError( f"recipe {self.id}: stage {stage.id} starts from the geometry of " f"{source!r}, which is not an earlier stage" ) if not seen[source].yields_geometry(): raise ValueError( f"recipe {self.id}: stage {stage.id} starts from the geometry of " f"{source!r}, a {seen[source].kind} stage, which yields none" ) if stage.kind == "hessian" and stage.hessian is not None and stage.parent: parent = seen[stage.geometry.stage or stage.parent] surface = parent.dft.surface if parent.dft is not None else None if stage.hessian.compare_parent_max_delta_cm is not None: comparison = seen[stage.parent] if ( comparison.hessian is None or comparison.hessian.surface != stage.hessian.surface or comparison.geometry != stage.geometry ): raise ValueError( "Hessian comparison needs a same-surface, same-geometry parent" ) if surface is not None and surface != stage.hessian.surface: raise ValueError( f"recipe {self.id}: the Hessian {stage.id} must run on the surface " f"of its parent optimisation {parent.id}, and it does not" ) seen[stage.id] = stage if len(set(self.species)) != len(self.species): raise ValueError(f"recipe {self.id}: a species is named twice") campaigns = [c.id for c in self.campaigns] if len(set(campaigns)) != len(campaigns): raise ValueError(f"recipe {self.id}: a campaign id is used twice") return self def stage_by_id(self, identifier: str) -> Stage: for stage in self.stages: if stage.id == identifier: return stage raise KeyError(identifier)
[docs] class Style(_Strict): """How exported figures look.""" preset: _Token = Field(default="acs-1996", description="A journal style preset.")
[docs] class Edit(_Strict): """One revision of the document: who changed it, through what, and why.""" revision: int = Field(ge=1) at: AwareDatetime actor: Literal["human", "agent", "system"] client: _Token | None = Field( default=None, description="The surface the edit came through: browser, cli, rest, mcp." ) summary: _Text
def _now() -> datetime: return datetime.now(UTC).replace(microsecond=0)
[docs] class Provenance(_Strict): """Where the document came from.""" app: _Text = Field(description="The application and version that created it.") created: AwareDatetime = Field(default_factory=_now) edits: list[Edit] = Field(default_factory=list)
#: Okabe and Ito's colour-universal-design palette, by name (see vibesketch.document.highlights). HighlightColour = Literal["orange", "sky-blue", "green", "blue", "vermillion", "purple"] class Highlight(_Strict): """Atoms of a species marked in a colour, in every drawing and figure of it. Atoms are canonical ranks of the species' heavy atoms (RDKit's ``CanonicalRankAtoms``), so the mark follows the structure, not the order it was drawn in; bonds between two marked atoms are marked too. """ species: _Identifier atoms: list[Annotated[int, Field(ge=0)]] = Field(min_length=1) colour: HighlightColour = "orange" @field_validator("atoms") @classmethod def _distinct(cls, atoms: list[int]) -> list[int]: if len(atoms) != len(set(atoms)): raise ValueError("an atom is marked twice in one highlight") return sorted(atoms)
[docs] class SurfaceAnchor(_Strict): """A schematic attachment point, fractions of the image box; y runs down.""" u: float = Field(default=0.5, ge=0, le=1) v: float = Field(default=0, ge=0, le=1) dx: float = 0 dy: float = 1 site: Literal["top", "bridge", "hollow"] | None = None repeat: int = Field(default=1, ge=1, le=10000) generics: list[_Identifier] = Field(default_factory=list) @model_validator(mode="after") def _direction(self) -> Self: if not all(math.isfinite(x) for x in (self.u, self.v, self.dx, self.dy)): raise ValueError("surface anchor coordinates must be finite") if math.hypot(self.dx, self.dy) < 1e-8: raise ValueError("surface anchor needs an outward direction") return self
[docs] class Surface(_Strict): """A schematic surface, not a physical atomistic or periodic model (#140).""" id: _Identifier kind: Literal["surface", "electrode", "stack", "nanoparticle", "nanotube-array"] = "surface" material: str = Field(min_length=1, max_length=80, pattern=r"^[A-Za-z0-9 ()/.,+-]+$") facet: str | None = Field(default=None, max_length=40, pattern=r"^[0-9 ()-]+$") layers: list[str] = Field(default_factory=list, max_length=12) diameter_nm: float | None = Field(default=None, gt=0, le=100000, allow_inf_nan=False) ligand_shell: str | None = Field(default=None, max_length=80, pattern=r"^[A-Za-z0-9 ()/.,+-]+$") nanotube: _Identifier | None = None count: int = Field(default=5, ge=1, le=20) anchors: dict[_Identifier, SurfaceAnchor] = Field( default_factory=lambda: {"top": SurfaceAnchor()}, max_length=32 ) @model_validator(mode="after") def _meaning(self) -> Self: import re if any(not re.fullmatch(r"[A-Za-z0-9 ()/.,+-]{1,80}", layer) for layer in self.layers): raise ValueError("surface layers need plain material labels") if self.kind == "stack" and not self.layers: raise ValueError("a substrate stack needs layers, bottom to top") if self.kind == "nanoparticle" and self.diameter_nm is None: raise ValueError("a nanoparticle needs its diameter_nm") if (self.kind == "nanotube-array") != (self.nanotube is not None): raise ValueError("only a nanotube-array requires a nanotube species") return self
[docs] class Document(_Strict): """A vibe-sketch document, schema version 1.""" model_config = ConfigDict( extra="forbid", populate_by_name=True, validate_assignment=True, title="vibe-sketch document", json_schema_extra={"description": "A .vsk file: a drawing and what to compute from it."}, ) schema_id: Literal["vibe-sketch/1"] = Field( default=SCHEMA_ID, alias="schema", description="The document format and its major version." ) revision: int = Field( default=0, ge=0, description="Incremented by the server on every accepted edit." ) canvas: Canvas = Field(default_factory=lambda: Canvas(format="ket")) species: list[Species] = Field(default_factory=list) reactions: list[Reaction] = Field(default_factory=list) jobs: list[Job] = Field(default_factory=list) recipes: list[Recipe] = Field( default_factory=list, description="Saved staged workflows over the document's species." ) highlights: list[Highlight] = Field( default_factory=list, description="Atoms marked in colour for figures and the canvas." ) generics: list[Generic] = Field( default_factory=list, description="R-group cores and bracketed groups the drawing shows; derived from it.", ) rgroups: list[RGroup] = Field( default_factory=list, description="The R-groups' definitions; derived from the drawing." ) surfaces: list[Surface] = Field( default_factory=list, description="Schematic surfaces derived from canvas metadata; not 3D models.", ) style: Style = Field(default_factory=Style) provenance: Provenance @model_validator(mode="after") def _references_resolve(self) -> Self: ids: list[str] = [ *(s.id for s in self.species), *(r.id for r in self.reactions), *(j.id for j in self.jobs), *(r.id for r in self.recipes), *(g.id for g in self.generics), *(s.id for s in self.surfaces), ] duplicates = sorted({i for i in ids if ids.count(i) > 1}) if duplicates: raise ValueError(f"identifiers used more than once: {', '.join(duplicates)}") numbers = [r.number for r in self.rgroups] if len(set(numbers)) != len(numbers): twice = sorted({f"R{n}" for n in numbers if numbers.count(n) > 1}) raise ValueError(f"R-groups defined more than once: {', '.join(twice)}") species = {s.id for s in self.species} generic_ids = {g.id for g in self.generics} for surface in self.surfaces: if surface.nanotube is not None: tube = next((s for s in self.species if s.id == surface.nanotube), None) if ( tube is None or tube.nanostructure is None or not isinstance(tube.nanostructure.parameters, Nanotube) ): raise ValueError(f"surface {surface.id} needs an existing nanotube species") if any(g not in generic_ids for a in surface.anchors.values() for g in a.generics): raise ValueError(f"surface {surface.id} names a missing generic attachment") for reaction in self.reactions: missing = [i for i in reaction.species_ids() if i not in species] if missing: raise ValueError( f"reaction {reaction.id} names species that do not exist: {', '.join(missing)}" ) for job in self.jobs: if job.species not in species: raise ValueError( f"job {job.id} names species {job.species!r}, which does not exist" ) for recipe in self.recipes: missing = [i for i in recipe.species if i not in species] if missing: raise ValueError( f"recipe {recipe.id} names species that do not exist: {', '.join(missing)}" ) for highlight in self.highlights: if highlight.species not in species: raise ValueError( f"a highlight names species {highlight.species!r}, which does not exist" ) return self
[docs] def species_by_id(self, identifier: str) -> Species: """The species with ``identifier``; raises ``KeyError`` when there is none.""" for species in self.species: if species.id == identifier: return species raise KeyError(identifier)
[docs] def job_by_id(self, identifier: str) -> Job: """The job with ``identifier``; raises ``KeyError`` when there is none.""" for job in self.jobs: if job.id == identifier: return job raise KeyError(identifier)
[docs] def to_json_data(self) -> dict[str, Any]: """The document as JSON-compatible data, in the field order of the schema. Unset optional fields are left out; the canvas data is kept exactly as it came, nulls included. """ data = self.model_dump(mode="json", by_alias=True, exclude_none=True) data["canvas"]["data"] = self.canvas.model_dump(mode="json")["data"] if not data.get("recipes"): data.pop("recipes", None) # additive: a document without recipes reads as before for reaction in data.get("reactions", []): if not reaction.get("amounts"): reaction.pop("amounts", None) # likewise for a reaction without amounts for field in ("coefficients", "states", "quantities"): if not reaction.get(field): reaction.pop(field, None) # additive (#141) for field in ("highlights", "generics", "rgroups", "surfaces"): if not data.get(field): data.pop(field, None) for generic in data.get("generics", []): if not generic.get("labels"): generic.pop("labels", None) # additive: only labelled structures have labels return data