Source code for rxn_network.flows.core

"""Core flows for the reaction-network package."""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import TYPE_CHECKING

from jobflow import Flow, Maker
from pymatgen.core.composition import Element

from rxn_network.core import Composition
from rxn_network.enumerators.basic import BasicEnumerator, BasicOpenEnumerator
from rxn_network.enumerators.minimize import MinimizeGibbsEnumerator, MinimizeGrandPotentialEnumerator
from rxn_network.jobs.core import (
    CalculateCompetitionMaker,
    GetEntrySetMaker,
    NetworkMaker,
    PathwaySolverMaker,
    ReactionEnumerationMaker,
)
from rxn_network.utils.funcs import get_logger

if TYPE_CHECKING:
    from collections.abc import Collection, Iterable

    from rxn_network.entries.entry_set import GibbsEntrySet

logger = get_logger(__name__)


[docs] @dataclass class SynthesisPlanningFlowMaker(Maker): """Maker to create an inorganic synthesis planning workflow. This flow has three stages. Steps: 1) Entries are acquired via `GetEntrySetMaker`. This job both gets the computed entries from a databse (e.g., Materials Project) and processes them for use in the reaction network. 2) Reactions are enumerated via the provided `ReactionEnumerationMaker` (and associated enumerators). This computes the full reaction network so that selectivities can be calculated. 3) The competition of all synthesis reactions to the desired target is assessed via the `CalculateCompetitionMaker`. This flow also has the option to include an "open" element and a list of chempots. This will enumerate reactions at different conditions and evaluate their selectivities at those conditinons. This flow does not produce a specific output document. Instead, it is convenient to analyze output documents from each of the jobs in the flow based on the desired analysis. For the final "results", one should access the reaction set produced by the `CalculateCompetitionMaker` at the conditions of interest. If you use this code in your work, please cite the following work: McDermott, M. J. et al. Assessing Thermodynamic Selectivity of Solid-State Reactions for the Predictive Synthesis of Inorganic Materials. ACS Cent. Sci. (2023) doi:10.1021/acscentsci.3c01051. Args: name: Name of the flow. Automatically generated if not provided. get_entry_set_maker: `GetEntrySetMaker`used to create the job for acquiring entries. Automatically generated with default settings if not provided. enumeration_maker: `ReactionEnumerationMaker` used to create the reaction enumeration job. Automatically generated with default settings if not provided. calculate_competition_maker: `CalculateCompetitionMaker` used to create the selectivity analysis job. Automatically generated with default settings if not provided. open_elem: Optional element to use as the "open" element. If provided, the flow will enumerate reactions at different chemical potentials of this element. chempots: List of chemical potentials to use for the "open" element. If provided, the flow will enumerate reactions at different chemical potentials of this element. use_basic_enumerators: Whether to use the `BasicEnumerator` and `BasicOpenEnumerator` enumerators in the enumeration job. use_minimize_enumerators: Whether to use the `MinimizeGibbsEnumerator` and the `MinimizeGrandPotentialEnumerator` enumerators in the enumeration job. basic_enumerator_kwargs: Keyword arguments to pass to the basic enumerators. minimize_enumerator_kwargs: Keyword arguments to pass to the minimize enumerators. """ name: str = "synthesis_planning" get_entry_set_maker: GetEntrySetMaker = field(default_factory=GetEntrySetMaker) enumeration_maker: ReactionEnumerationMaker = field(default_factory=ReactionEnumerationMaker) calculate_competition_maker: CalculateCompetitionMaker = field(default_factory=CalculateCompetitionMaker) open_elem: Element | str | None = None chempots: list[float] | None = None use_basic_enumerators: bool = True use_minimize_enumerators: bool = True basic_enumerator_kwargs: dict = field(default_factory=dict) minimize_enumerator_kwargs: dict = field(default_factory=dict) def __post_init__(self): self.open_elem = Element(self.open_elem) if self.open_elem else None self.open_formula = Composition(str(self.open_elem)).reduced_formula if self.open_elem else None
[docs] def make( # type: ignore self, target_formula: str, added_elems: Collection[str] | None = None, entries: GibbsEntrySet | None = None, ): """Returns a flow used for planning optimal synthesis recipes to a specified target. Args: target_formula: The chemical formula of a target phase (e.g., "BaTiO3"). added_elems: An optional list of additional elements to consider (e.g., ["C", "H"]). Defaults to None. entries: An optional provided set of entries to enumerate from. If entries are not provided, then they will be acquired from a database (e.g., Materials Project) and processed using the GetEntrySetMaker. """ target_formula = Composition(target_formula).reduced_formula flow_name = f"Synthesis planning: {target_formula}" if added_elems is None: added_elems = [] else: flow_name = flow_name + f" (+ {'-'.join(sorted(added_elems))})" flow_name = flow_name + f", T={self.get_entry_set_maker.temperature} K" chemsys = "-".join( sorted({str(e) for e in Composition(target_formula).elements} | {str(e) for e in added_elems}) ) jobs = [] if entries is None: get_entry_set_maker = self.get_entry_set_maker.update_kwargs( { "name": self.get_entry_set_maker.name + f" ({chemsys}, T={self.get_entry_set_maker.temperature} K," f" +{round(self.get_entry_set_maker.e_above_hull, 3)} eV)", "formulas_to_include": list({*self.get_entry_set_maker.formulas_to_include, target_formula}), } ) get_entry_set_job = get_entry_set_maker.make(chemsys) jobs.append(get_entry_set_job) entries = get_entry_set_job.output.entries targets = [target_formula] filter_by_chemsys = Composition(target_formula).chemical_system basic_enumerator_kwargs = self.basic_enumerator_kwargs.copy() minimize_enumerator_kwargs = self.minimize_enumerator_kwargs.copy() kwarg_update = {"targets": targets, "filter_by_chemsys": filter_by_chemsys} basic_enumerator_kwargs.update(kwarg_update) minimize_enumerator_kwargs.update(kwarg_update) enumerators = [] if self.use_basic_enumerators: enumerators.append( BasicEnumerator( filter_by_chemsys=filter_by_chemsys, **self.basic_enumerator_kwargs, ) ) if self.use_minimize_enumerators: enumerators.append( MinimizeGibbsEnumerator( filter_by_chemsys=filter_by_chemsys, **self.minimize_enumerator_kwargs, ) ) enumeration_job = self.enumeration_maker.make(enumerators=enumerators, entries=entries) jobs.append(enumeration_job) base_rxn_set = enumeration_job.output.rxns calculate_competition_maker = self.calculate_competition_maker base_calculate_competition_job = calculate_competition_maker.make( rxn_sets=[base_rxn_set], entries=entries, target_formula=target_formula, ) jobs.append(base_calculate_competition_job) if self.open_elem and self.chempots: for chempot in self.chempots: subname = f"(open {self.open_elem!s}, mu={chempot})" enumeration_maker = self.enumeration_maker.update_kwargs( {"name": self.enumeration_maker.name + subname}, nested=False, ) calculate_competition_maker = calculate_competition_maker.update_kwargs( { "chempot": chempot, "open_elem": self.open_elem, "name": self.calculate_competition_maker.name + subname, }, nested=False, ) open_enumerators = [] if self.use_basic_enumerators: open_enumerators.append( BasicOpenEnumerator( open_phases=[self.open_formula], **self.basic_enumerator_kwargs, ) ) if self.use_minimize_enumerators: open_enumerators.append( MinimizeGrandPotentialEnumerator( open_elem=self.open_elem, mu=chempot, filter_by_chemsys=filter_by_chemsys, **self.minimize_enumerator_kwargs, ) ) enumeration_job = enumeration_maker.make( enumerators=open_enumerators, entries=entries, ) jobs.append(enumeration_job) calculate_competition_job = calculate_competition_maker.make( rxn_sets=[base_rxn_set, enumeration_job.output.rxns], entries=entries, target_formula=target_formula, ) jobs.append(calculate_competition_job) return Flow(jobs, name=flow_name)
[docs] @dataclass class NetworkFlowMaker(Maker): """Maker to create a chemical reaction network and perform (balanced) pathfinding on the network. This flow has four stages: 1) Entries are acquired via `GetEntrySetMaker`. This job both gets the computed entries from a databse (e.g., Materials Project) and processes them for use in the reaction network. 2) Reactions are enumerated via the provided `ReactionEnumerationMaker` (and associated enumerators). 3) The network is created using `NetworkMaker` and basic paths are found (k-shortest paths to each target). 4) The final balanced reaction pathways are produced using the `SolverMaker`. If you use this code in your own work, please consider citing this paper: McDermott, M. J.; Dwaraknath, S. S.; Persson, K. A. A Graph-Based Network for Predicting Chemical Reaction Pathways in Solid-State Materials Synthesis. Nature Communications 2021, 12 (1), 3097. https://doi.org/10.1038/s41467-021-23339-x. Args: name: The name of the network flow. Automatically assigned if not provided. get_entry_set_maker: `GetEntrySetMaker`used to create the job for acquiring entries. Automatically generated with default settings if not provided. enumeration_maker: `ReactionEnumerationMaker` used to create the reaction enumeration job. Automatically generated with default settings if not provided. network_maker: `NetworkMaker` used to create the reaction network from sets of reactions. Also identifies basic reaction pathways. Automatically generated with default settings if not provided. solver_maker: `PathwaySolverMaker` used to find balanced reaction pathways from set of pathways emerging from pathfinding. Automatically generated with default settings if not provided. open_elem: Optional element to use as the "open" element. If provided, the flow will enumerate reactions at different chemical potentials of this element. chempots: List of chemical potentials to use for the "open" element. If provided, the flow will enumerate reactions at different chemical potentials of this element. use_basic_enumerators: Whether to use the `BasicEnumerator` and `BasicOpenEnumerator` enumerators in the enumeration job. use_minimize_enumerators: Whether to use the `MinimizeGibbsEnumerator` and the `MinimizeGrandPotentialEnumerator` enumerators in the enumeration job. basic_enumerator_kwargs: Keyword arguments to pass to the basic enumerators. minimize_enumerator_kwargs: Keyword arguments to pass to the minimize enumerators. """ name: str = "find_reaction_pathways" get_entry_set_maker: GetEntrySetMaker = field(default_factory=GetEntrySetMaker) enumeration_maker: ReactionEnumerationMaker = field(default_factory=ReactionEnumerationMaker) network_maker: NetworkMaker = field(default_factory=NetworkMaker) solver_maker: PathwaySolverMaker | None = None open_elem: Element | None = None chempots: list[float] | None = None use_basic_enumerators: bool = True use_minimize_enumerators: bool = True basic_enumerator_kwargs: dict = field(default_factory=dict) minimize_enumerator_kwargs: dict = field(default_factory=dict) def __post_init__(self): self.open_elem = Element(self.open_elem) if self.open_elem else None self.open_formula = Composition(str(self.open_elem)).reduced_formula if self.open_elem else None
[docs] def make(self, precursors: Iterable[str], targets: Iterable[str], entries: GibbsEntrySet | None = None): """Returns a flow used for finding reaction pathways between precursors and targets. Args: precursors: precursor formulas targets: target formulas entries: Optional entry set. If not provided, entries will be automatically acquired from Materials Project. Defaults to None. Returns: _description_ """ precursor_formulas = [Composition(f).reduced_formula for f in precursors] target_formulas = [Composition(f).reduced_formula for f in targets] flow_name = ( f"Reaction network analysis: {'-'.join(sorted(precursor_formulas))} ->" f" {'-'.join(sorted(target_formulas))}" ) chemsys = "-".join( {str(e) for formula in precursor_formulas + target_formulas for e in Composition(formula).elements} ) jobs = [] if entries is None: get_entry_set_maker = self.get_entry_set_maker.update_kwargs( { "name": self.get_entry_set_maker.name + f" ({chemsys}, T={self.get_entry_set_maker.temperature} K," f" +{round(self.get_entry_set_maker.e_above_hull, 3)} eV)", "formulas_to_include": list( set(self.get_entry_set_maker.formulas_to_include + precursor_formulas + target_formulas) ), } ) get_entry_set_job = get_entry_set_maker.make(chemsys) jobs.append(get_entry_set_job) entries = get_entry_set_job.output.entries self.basic_enumerator_kwargs.copy() self.minimize_enumerator_kwargs.copy() enumerators = [] if self.use_basic_enumerators: enumerators.append( BasicEnumerator( **self.basic_enumerator_kwargs, ) ) if self.open_formula: enumerators.append( BasicOpenEnumerator( open_phases=[self.open_formula], **self.basic_enumerator_kwargs, ) ) if self.use_minimize_enumerators: enumerators.append( MinimizeGibbsEnumerator( **self.minimize_enumerator_kwargs, ) ) enumeration_job = self.enumeration_maker.make(enumerators=enumerators, entries=entries) jobs.append(enumeration_job) base_rxn_set = enumeration_job.output.rxns base_network_job = self.network_maker.make([base_rxn_set]) jobs.append(base_network_job) if self.solver_maker: base_pathway_job = self.solver_maker.make( base_network_job.output.paths, entries=base_network_job.output.network.entries, ) jobs.append(base_pathway_job) if self.use_minimize_enumerators and self.open_elem and self.chempots: for chempot in self.chempots: subname = f"(open {self.open_elem!s}, mu={chempot})" enumeration_maker = self.enumeration_maker.update_kwargs( {"name": self.enumeration_maker.name + subname}, nested=False, ) network_maker = self.network_maker.update_kwargs( { "name": self.network_maker.name + subname, "chempot": chempot, "open_elem": self.open_elem, }, nested=False, ) if self.solver_maker: solver_maker = self.solver_maker.update_kwargs( { "name": self.solver_maker.name + subname, "chempot": chempot, "open_elem": self.open_elem, }, nested=False, ) enumerator = MinimizeGrandPotentialEnumerator( open_elem=self.open_elem, mu=chempot, **self.minimize_enumerator_kwargs, ) enumeration_job = enumeration_maker.make(enumerators=[enumerator], entries=entries) network_job = network_maker.make([base_rxn_set, enumeration_job.output.rxns]) jobs.extend([enumeration_job, network_job]) if self.solver_maker: pathway_job = solver_maker.make( network_job.output.paths, entries=network_job.output.network.entries, ) jobs.append(pathway_job) return Flow(jobs, name=flow_name)