Spaces:
Build error
Build error
import os | |
import re | |
from langchain.agents import Tool, tool | |
# from mp_api.client import MPRester | |
from pymatgen.ext.matproj import MPRester | |
from rxn_network.entries.entry_set import GibbsEntrySet | |
from rxn_network.enumerators.basic import BasicEnumerator | |
class SynthesisReactions: | |
def __init__(self, temp=900, stabl=0.025, exclusive_precursors=False, exclusive_targets=False): | |
self.temp = temp | |
self.stabl = stabl | |
self.exclusive_precursors = exclusive_precursors | |
self.exclusive_targets = exclusive_targets | |
def _split_string(self, s): | |
if isinstance(s, list): | |
s = "".join(s) | |
parts = re.findall('[a-z]+|[A-Z][a-z]*', s) | |
letters_only = [re.sub(r'\d+', '', part) for part in parts] | |
unique_letters = list(set(letters_only)) | |
result = "-".join(unique_letters) | |
return result | |
def _get_rxn_from_precursor(self, precursors_formulas): | |
prec = precursors_formulas.split(',') if "," in precursors_formulas else precursors_formulas | |
with MPRester(os.getenv("MAPI_API_KEY")) as mpr: | |
entries = mpr.get_entries_in_chemsys(self._split_string(prec)) | |
gibbs_entries = GibbsEntrySet.from_computed_entries(entries, self.temp) | |
filtered_entries = gibbs_entries.filter_by_stability(self.stabl) | |
prec = [prec] if isinstance(prec, str) else prec | |
be = BasicEnumerator(precursors=prec, exclusive_precursors=self.exclusive_precursors) | |
rxns = be.enumerate(filtered_entries) | |
try: | |
rxn_choice = next(iter(rxns)) | |
return str(rxn_choice) | |
except: | |
return "Error: No reactions found." | |
def _get_rxn_from_target(self, targets_formulas): | |
targets = targets_formulas.split(',') if "," in targets_formulas else targets_formulas | |
with MPRester(os.getenv("MAPI_API_KEY")) as mpr: | |
entries = mpr.get_entries_in_chemsys(self._split_string(targets)) | |
gibbs_entries = GibbsEntrySet.from_computed_entries(entries, self.temp) | |
filtered_entries = gibbs_entries.filter_by_stability(self.stabl) | |
targets = [targets] if isinstance(targets, str) else targets | |
be = BasicEnumerator(targets=targets, exclusive_targets=self.exclusive_targets) | |
rxns = be.enumerate(filtered_entries) | |
try: | |
rxn_choice = next(iter(rxns)) | |
return str(rxn_choice) | |
except: | |
return "Error: No reactions found." | |
def _break_equation(self, equation): | |
pattern = r'(\d*\.?\d*\s*[A-Za-z]+\d*|\+|\->)' | |
pieces = re.findall(pattern, equation) | |
equation_pieces = [] | |
current_piece = '' | |
for piece in pieces: | |
if piece == '+' or piece == '->': | |
equation_pieces.append(current_piece.strip()) | |
equation_pieces.append(piece) | |
current_piece = '' | |
else: | |
current_piece += piece + ' ' | |
equation_pieces.append(current_piece.strip()) | |
return equation_pieces | |
def _convert_equation_pieces(self, equation_pieces): | |
if '+' in equation_pieces: | |
equation_pieces = [piece if piece != '+' else 'with' for piece in equation_pieces] | |
equation_pieces = [piece if piece != '->' else 'to yield' for piece in equation_pieces] | |
else: | |
equation_pieces = [piece if piece != '->' else 'yields' for piece in equation_pieces] | |
return equation_pieces | |
def _split_equation_pieces(self, equation_pieces): | |
new_pieces = [] | |
for piece in equation_pieces: | |
if piece in ["with", "to yield", "yields"]: | |
new_pieces.append(piece) | |
else: | |
if re.match(r'^\d*\.\d+|\d+', piece): | |
number_match = re.match(r'^\d*\.\d+|\d+', piece) | |
number = number_match.group(0) | |
rest = piece[len(number):] | |
new_pieces.append(number) | |
new_pieces.append(rest) | |
else: | |
new_pieces.append("1") | |
new_pieces.append(piece) | |
return new_pieces | |
def _modify_mols(self, equation_pieces): | |
for i, piece in enumerate(equation_pieces): | |
if piece.replace('.', '', 1).isdigit(): | |
equation_pieces[i] = f"{piece} mols" | |
return equation_pieces | |
def _combine_equation_pieces(self, equation_pieces): | |
if 'with' in equation_pieces: | |
equation_pieces.insert(0, 'mix') | |
combined_string = ' '.join(equation_pieces) | |
return combined_string | |
def _process_equation(self, equation): | |
equation_pieces = self._break_equation(equation) | |
converted_pieces = self._convert_equation_pieces(equation_pieces) | |
split_pieces = self._split_equation_pieces(converted_pieces) | |
modified_pieces = self._modify_mols(split_pieces) | |
combined_string = self._combine_equation_pieces(modified_pieces) | |
return combined_string | |
def get_reaction(self, input_string): | |
input_parts = input_string.split(',', 1) | |
if len(input_parts) != 2: | |
raise ValueError("Invalid input format. Expected 'precursor' or 'target', followed by a comma, and then the list of formulas separated by a comma.") | |
mode, formulas = input_parts | |
mode = mode.lower().strip() | |
if mode == "precursor": | |
reaction = self._get_rxn_from_precursor(formulas) | |
elif mode == "target": | |
reaction = self._get_rxn_from_target(formulas) | |
else: | |
raise ValueError("Invalid mode. Expected 'precursor' or 'target'.") | |
processed_reaction = self._process_equation(reaction) | |
return processed_reaction | |
def get_tools(self): | |
return [ | |
Tool( | |
name = "Get a synthesis reaction for a material", | |
func = self.get_reaction, | |
description = ( | |
"This function is useful for suggesting a synthesis reaction for a material. " | |
"Give this tool a string containing either precursor or target, then a comma, followed by the formulas separated by comma as input and returns a synthesis reaction." | |
"The mode is used to determine if the input is a precursor or a target material. " | |
) | |
)] | |