Source code for rxn_network.jobs.schema

"""Core definition for various task and synthesis recipe documents."""

from typing import Any, Optional

from pydantic import BaseModel, Field
from pymatgen.core.composition import Element

from rxn_network.entries.entry_set import GibbsEntrySet
from rxn_network.enumerators.base import Enumerator
from rxn_network.network.base import Network
from rxn_network.pathways.pathway_set import PathwaySet
from rxn_network.pathways.solver import Solver
from rxn_network.reactions.computed import ComputedReaction
from rxn_network.reactions.reaction_set import ReactionSet
from rxn_network.utils.funcs import datetime_str


[docs] class EntrySetDocument(BaseModel): """A single entry set object as produced by the GetEntrySet job.""" task_label: Optional[str] = Field(None, description="The name of the task.") last_updated: str = Field( default_factory=datetime_str, description="Timestamp of when the document was last updated.", ) entries: GibbsEntrySet = Field(description="The entry set object.") e_above_hull: Optional[float] = Field(None, description="The e_above_hull cutoff.") include_polymorphs: bool = Field( default=False, description="Whether to include metastable polymorphs in the entry set." ) formulas_to_include: Optional[list[str]] = Field( None, description="The required formulas to include during construciton." )
[docs] class EnumeratorTaskDocument(BaseModel): """A single task object from the reaction enumerator workflow.""" task_label: Optional[str] = Field(None, description="The name of the task.") last_updated: str = Field( default_factory=datetime_str, description="Timestamp of when the document was last updated.", ) rxns: ReactionSet = Field(description="The reaction set.") targets: Optional[list[str]] = Field(None, description="The target formulas used in the enumerator(s).") elements: Optional[list[Element]] = Field(None, description="The elements of the total chemical system") chemsys: Optional[str] = Field(None, description="The total chemical system string (e.g., Fe-Li-O).") added_elements: Optional[list[Element]] = Field( None, description="The elements added beyond the elements of the target(s)." ) added_chemsys: Optional[str] = Field(None, description="The chemical system of the added elements") enumerators: Optional[list[Enumerator]] = Field( None, description="A list of the enumerator objects used to calculate the reactions.", )
[docs] class CompetitionTaskDocument(BaseModel): """A document containing the reactions and their selectivities as created by the CalculateCompetitionMaker job. """ task_label: Optional[str] = Field(None, description="The name of the task.") last_updated: str = Field( default_factory=datetime_str, description="Timestamp of when the document was last updated.", ) rxns: ReactionSet = Field( description=( "The reaction set, where thereactions have calculated competition" " information stored in their data attribute." ) ) target_formula: str = Field(description="The reduced chemical formula of the target material.") open_elem: Optional[Element] = Field(None, description="The open element (if any).") chempot: Optional[float] = Field(None, description="The chemical potential of the open element.") added_elements: Optional[list[Element]] = Field( None, description="The elements added beyond the elements of the target(s)." ) added_chemsys: Optional[str] = Field(None, description="The chemical system of the added elements.") calculate_competition: Optional[bool] = Field(None, description="Whether to calculate competition scores.") calculate_chempot_distances: Optional[bool] = Field( None, description="Whether to calculate chemical potential distances." ) temp: Optional[float] = Field( None, description=("The temperature in K used to determine the primary competition weightings."), ) batch_size: Optional[int] = Field(None, description="The batch size for the reaction set") cpd_kwargs: Optional[dict[str, Any]] = Field(None, description="The kwargs for ChempotDistanceCalculator.")
[docs] class NetworkTaskDocument(BaseModel): """The calculation output from the NetworkMaker workflow. Optionally includes unbalanced paths found during pathfinding. """ task_label: Optional[str] = Field(None, description="The name of the task.") last_updated: str = Field( default_factory=datetime_str, description="Timestamp of when the document was last updated.", ) network: Network = Field(description="The reaction network") paths: Optional[PathwaySet] = Field(None, description="The (simple/unbalanced) reaction pathways") k: Optional[int] = Field(None, description="The number of paths solved for, if any.") precursors: Optional[list[str]] = Field(None, description="The precursor formulas.") targets: Optional[list[str]] = Field(None, description="The target formulas.")
[docs] class PathwaySolverTaskDocument(BaseModel): """The calculation output from the PathwaySolverMaker workflow. Contains the pathway solver object and its calculated balanced pathways. """ task_label: Optional[str] = Field(None, description="The name of the task.") last_updated: str = Field( default_factory=datetime_str, description="Timestamp of when the document was last updated.", ) solver: Solver = Field(description="The pathway solver used to calculate pathways.") balanced_paths: PathwaySet = Field(description="The balanced reaction pathways.") precursors: list[str] = Field(description="The precursor compositions.") targets: list[str] = Field(description="The target compositions.") net_rxn: ComputedReaction = Field(description="The net reaction used for pathway solving.") max_num_combos: int = Field(description=("The maximum number of combinations to consider in the pathway solver.")) find_intermediate_rxns: bool = Field(description="Whether to find reactions from intermediate compositions.") intermediate_rxn_energy_cutoff: float = Field( description="The mximum energy cutoff for filtering intermediate reactions." ) use_basic_enumerator: bool = Field(description="Whether to use the basic enumerators in path solving.") use_minimize_enumerator: bool = Field(description="Whether to use the minimize enumerators in path solving.") filter_interdependent: bool = Field(description="Whether to filter out interdependent pathway.s")