diff --git a/arc/family/family.py b/arc/family/family.py index ad474564a3..071ae23385 100644 --- a/arc/family/family.py +++ b/arc/family/family.py @@ -1453,17 +1453,58 @@ def isomorphic_products(rxn: ARCReaction, return check_product_isomorphism(products, p_species) -def check_family_name(family: str +def check_family_name(family: str, + rmg_family_set: list[str] | str | None = None, ) -> bool: """ Check whether the family name is defined. Args: family (str): The family name. + rmg_family_set (list[str] | str, optional): The RMG family set to look the family up in. + ``None`` (the default) means ``settings['rmg_family_set']``, + read on every call. Pass ``'all'`` to ask whether the family + is available at all rather than whether it is configured. Returns: bool: Whether the family is defined. """ - if not isinstance(family, str) and family is not None: + if family is None: + return True + if not isinstance(family, str): raise TypeError("Family name must be a string or None.") - return family in get_all_families() or family is None + return family in get_all_families(rmg_family_set=rmg_family_set) + + +def is_family_available(family: str) -> bool: + """ + Check whether a reaction family can be loaded by label, whether or not the configured family set + lists it. A family is available when ``get_all_families('all')`` reaches it, or when the + configured ``settings['rmg_family_set']`` does. The second term is what keeps a surface family + available, since ``'all'`` skips the surface family sets and directories. + + Args: + family (str): The family name. + + Returns: + bool: Whether the family can be loaded by label. + """ + return check_family_name(family, rmg_family_set='all') or check_family_name(family) + + +def get_families_from_product_dicts(product_dicts: list[dict]) -> list[str]: + """ + List the reaction families represented in a list of family product dictionaries, + without repetitions and in the order in which they first appear. + + Args: + product_dicts (list[dict]): The family product dictionaries to read. + + Returns: + list[str]: The family labels. + """ + families = list() + for product_dict in product_dicts: + if product_dict['family'] not in families: + families.append(product_dict['family']) + return families diff --git a/arc/family/family_test.py b/arc/family/family_test.py index 1aed269ff6..821a4bacf2 100644 --- a/arc/family/family_test.py +++ b/arc/family/family_test.py @@ -22,6 +22,8 @@ filter_products_by_reaction, get_reaction_family_products, get_all_families, + get_families_from_product_dicts, + is_family_available, get_rmg_family_directories, get_entries, split_entries, @@ -1810,6 +1812,39 @@ def test_check_family_name(self): with self.assertRaises(TypeError): check_family_name(123) + def test_check_family_name_against_a_named_family_set(self): + """Test that check_family_name() can ask whether a family is available rather than whether + the configured family set contains it. XY_Addition_MultipleBond is an available family that + only the halogens set lists, so the shipped 'default' set does not reach it.""" + with mock.patch.dict(settings, {'rmg_family_set': 'default'}): + self.assertFalse(check_family_name('XY_Addition_MultipleBond')) + self.assertTrue(check_family_name('XY_Addition_MultipleBond', rmg_family_set='all')) + self.assertTrue(check_family_name('XY_Addition_MultipleBond', rmg_family_set='halogens')) + self.assertTrue(check_family_name('ether_hydrolysis', rmg_family_set='all')) + self.assertFalse(check_family_name('Not_A_Family', rmg_family_set='all')) + self.assertTrue(check_family_name(None, rmg_family_set='all')) + + def test_is_family_available(self): + """Test that a family is available when 'all' reaches it or when the configured set does. + XY_Addition_MultipleBond is available although 'default' does not list it, and a surface + family is available only under a configured set that lists it, since 'all' skips them.""" + surface_family = get_rmg_recommended_family_sets()['surface'][0] + with mock.patch.dict(settings, {'rmg_family_set': 'default'}): + self.assertTrue(is_family_available('H_Abstraction')) + self.assertTrue(is_family_available('XY_Addition_MultipleBond')) + self.assertTrue(is_family_available('ether_hydrolysis')) + self.assertFalse(is_family_available('Not_A_Family')) + self.assertFalse(is_family_available(surface_family)) + with mock.patch.dict(settings, {'rmg_family_set': 'surface'}): + self.assertTrue(is_family_available(surface_family)) + self.assertTrue(is_family_available('H_Abstraction')) + + def test_get_families_from_product_dicts(self): + """Test listing the families of a product dicts list without repetitions and in order""" + self.assertEqual(get_families_from_product_dicts(list()), list()) + product_dicts = [{'family': 'H_Abstraction'}, {'family': 'Substitution_O'}, {'family': 'H_Abstraction'}] + self.assertEqual(get_families_from_product_dicts(product_dicts), ['H_Abstraction', 'Substitution_O']) + class TestFamilyChoiceGates(unittest.TestCase): """Unit tests for the non-lexical gates that choose between co-matching families""" diff --git a/arc/reaction/reaction.py b/arc/reaction/reaction.py index 226ffb5fc4..a2e93ac323 100644 --- a/arc/reaction/reaction.py +++ b/arc/reaction/reaction.py @@ -4,7 +4,11 @@ from arc.common import get_element_mass, get_logger from arc.exceptions import ReactionError, InputError -from arc.family.family import ReactionFamily, get_reaction_family_products, check_family_name +from arc.family.family import (ReactionFamily, + get_families_from_product_dicts, + get_reaction_family_products, + is_family_available, + ) from arc.molecule.resonance import generate_resonance_structures_safely from arc.species.converter import (check_xyz_dict, sort_xyz_using_indices, @@ -112,10 +116,11 @@ def __init__(self, self.long_kinetic_description = '' self._family = None self._family_determined = False + self._family_pinned = False self._family_own_reverse = None self._product_dicts = None if family is not None: - if check_family_name(family): + if is_family_available(family): self.family = family else: raise ValueError(f"Invalid family name: {family}") @@ -266,12 +271,14 @@ def family(self, value): """ Allow setting family. Product dictionaries that were already generated are restricted to the new family, and are discarded so that they are regenerated on demand if none of them belong - to it. + to it. A family set here is a pinned family: it becomes the family set that ``get_product_dicts()`` + loads, while a family that was determined lazily by the ``family`` property does not. """ if value is not None and not isinstance(value, str): raise InputError(f'Reaction family must be a string, got {value} which is a {type(value)}.') self._family = value self._family_determined = True + self._family_pinned = value is not None self._family_own_reverse = None if self._product_dicts is not None and value is not None: self._product_dicts = [product_dict for product_dict in self._product_dicts @@ -595,23 +602,54 @@ def get_product_dicts(self, 'discovered_in_reverse': bool: Whether the reaction was discovered in reverse}, ] + A family pinned on this reaction is used as the family set to generate from, so that a family + which the configured ``settings['rmg_family_set']`` does not contain still generates its + products. Both an RMG family and an ARC family are reached this way. A family the ``family`` + property determined lazily is not a pinned family and does not narrow anything. The narrowing + is applied only when no set is named at the call site and both family sources are considered. + Args: rmg_family_set (list[str] | str, optional): The RMG family set to use. - ``None`` (the default) means ``settings['rmg_family_set']``, - read on every call. + ``None`` (the default) means this reaction's family if one + is pinned, otherwise ``settings['rmg_family_set']``, read on + every call. consider_rmg_families (bool, optional): Whether to consider RMG's families in addition to ARC's. consider_arc_families (bool, optional): Whether to consider ARC's families in addition to RMG's. discover_own_reverse_rxns_in_reverse (bool, optional): Whether to discover own reverse reactions in reverse. + Raises: + ReactionError: If a family pinned on this reaction is either not an available family, + or generates no products for this reaction. + Returns: list[dict]: A list of dictionaries with the RMG reaction family products. """ + use_pinned_family = rmg_family_set is None and self._family_pinned \ + and consider_rmg_families and consider_arc_families + if use_pinned_family: + self.check_family() + rmg_family_set = [self._family] + logger.info(f'Reaction {self.label} was assigned the {self._family} family, ' + f'so only that family is searched for it.') product_dicts = get_reaction_family_products(rxn=self, rmg_family_set=rmg_family_set, consider_rmg_families=consider_rmg_families, consider_arc_families=consider_arc_families, discover_own_reverse_rxns_in_reverse=discover_own_reverse_rxns_in_reverse, ) + if use_pinned_family and not len(product_dicts): + families = get_families_from_product_dicts( + get_reaction_family_products(rxn=self, + rmg_family_set='all', + consider_rmg_families=consider_rmg_families, + consider_arc_families=consider_arc_families, + discover_own_reverse_rxns_in_reverse=discover_own_reverse_rxns_in_reverse, + )) + if len(families): + raise ReactionError(f'Reaction {self.label} was assigned the {self._family} family, but it does not ' + f'match this family. The families it does match are: {families}.') + raise ReactionError(f'Reaction {self.label} was assigned the {self._family} family, but it does not ' + f'match this family, nor any other available reaction family.') return product_dicts def restrict_product_dicts_to_family(self, @@ -635,10 +673,7 @@ def restrict_product_dicts_to_family(self, """ if not len(product_dicts): return product_dicts - families = list() - for product_dict in product_dicts: - if product_dict['family'] not in families: - families.append(product_dict['family']) + families = get_families_from_product_dicts(product_dicts) family = self._family if self._family is not None else families[0] restricted = [product_dict for product_dict in product_dicts if product_dict['family'] == family] if not len(restricted): @@ -680,16 +715,22 @@ def determine_family(self, """ Determine the RMG reaction family. When all arguments are left at their defaults, the cached ``product_dicts`` property is used - instead of generating a new product dicts list. + instead of generating a new product dicts list. Product dicts are otherwise generated through + ``get_product_dicts()``, so a family pinned on this reaction narrows what is loaded here too. Args: rmg_family_set (list[str] | str, optional): The RMG family set to use. - ``None`` (the default) means ``settings['rmg_family_set']``, - read on every call. + ``None`` (the default) means this reaction's family if one + is pinned, otherwise ``settings['rmg_family_set']``, read on + every call. consider_rmg_families (bool, optional): Whether to consider RMG's families in addition to ARC's. consider_arc_families (bool, optional): Whether to consider ARC's families in addition to RMG's. discover_own_reverse_rxns_in_reverse (bool, optional): Whether to discover own reverse reactions in reverse. + Raises: + ReactionError: If a family pinned on this reaction is either not an available family, + or generates no products for this reaction. + Returns: tuple[str | None, bool | None]: The reaction family label, and whether the family's template also represents its own reverse. @@ -697,20 +738,31 @@ def determine_family(self, if rmg_family_set is None and consider_rmg_families and consider_arc_families and not discover_own_reverse_rxns_in_reverse: product_dicts = self.product_dicts else: - product_dicts = get_reaction_family_products(rxn=self, - rmg_family_set=rmg_family_set, - consider_rmg_families=consider_rmg_families, - consider_arc_families=consider_arc_families, - discover_own_reverse_rxns_in_reverse=discover_own_reverse_rxns_in_reverse, - ) + product_dicts = self.get_product_dicts(rmg_family_set=rmg_family_set, + consider_rmg_families=consider_rmg_families, + consider_arc_families=consider_arc_families, + discover_own_reverse_rxns_in_reverse=discover_own_reverse_rxns_in_reverse, + ) if len(product_dicts): family, family_own_reverse = product_dicts[0]['family'], product_dicts[0]['own_reverse'] return family, family_own_reverse return None, None + def check_family(self): + """ + Check that a family pinned on this reaction can be loaded by label. + + Raises: + ReactionError: If the pinned family is not an available family. + """ + if self._family_pinned and not is_family_available(self._family): + raise ReactionError(f'Reaction {self.label} was assigned the {self._family} family, ' + f'which is not an available RMG or ARC reaction family.') + def check_attributes(self): """Check that the Reaction object is defined correctly""" self.set_label_reactants_products() + self.check_family() if not self.label: raise ReactionError('A reaction seems to not be defined correctly') if self.arrow not in self.label: diff --git a/arc/reaction/reaction_test.py b/arc/reaction/reaction_test.py index 9870e6b80e..8b4ffdad1b 100644 --- a/arc/reaction/reaction_test.py +++ b/arc/reaction/reaction_test.py @@ -14,7 +14,7 @@ from arc.common import ARC_PATH, ARC_TESTING_PATH, almost_equal_lists, read_yaml_file from arc.exceptions import ReactionError -from arc.family.family import get_all_families, get_rmg_recommended_family_sets +from arc.family.family import get_all_families, get_reaction_family_products, get_rmg_recommended_family_sets from arc.imports import settings from arc.main import ARC from arc.reaction.reaction import ARCReaction, remove_dup_species @@ -601,6 +601,177 @@ def test_pinning_a_family_that_does_not_match_raises(self): with self.assertRaises(ReactionError): _ = rxn.product_dicts + def test_a_pinned_family_is_the_family_set_that_is_loaded(self): + """XY_Addition_MultipleBond requires a halogen at *4, so RMG lists it in the halogens set and + the shipped 'default' set does not contain it. Pinning it makes it the family set that is + loaded, so C2H4 + HCl <=> C2H5Cl resolves to it without widening the configured setting.""" + def build_rxn(**kwargs): + return ARCReaction(r_species=[ARCSpecies(label='C2H4', smiles='C=C'), + ARCSpecies(label='HCl', smiles='Cl')], + p_species=[ARCSpecies(label='C2H5Cl', smiles='CCCl')], + **kwargs) + loaded_sets = list() + + def spy(**kwargs): + loaded_sets.append(kwargs.get('rmg_family_set')) + return get_reaction_family_products(**kwargs) + with mock.patch.dict(settings, {'rmg_family_set': 'default'}): + self.assertNotIn('XY_Addition_MultipleBond', get_all_families()) + self.assertEqual(build_rxn().product_dicts, list()) + self.assertIsNone(build_rxn().family) + with mock.patch('arc.reaction.reaction.get_reaction_family_products', side_effect=spy): + rxn = build_rxn(family='XY_Addition_MultipleBond') + product_dicts = rxn.product_dicts + self.assertEqual(rxn.determine_family(), ('XY_Addition_MultipleBond', False)) + self.assertGreater(len(product_dicts), 0) + self.assertEqual({product_dict['family'] for product_dict in product_dicts}, + {'XY_Addition_MultipleBond'}) + self.assertEqual(loaded_sets, [['XY_Addition_MultipleBond']]) + + def test_a_family_pinned_in_a_reaction_dictionary_is_the_family_set_that_is_loaded(self): + """A family pinned in the input file reaches the reaction through the reaction dictionary + rather than through the family argument, and narrows the loaded family set from there too.""" + r_species = [ARCSpecies(label='C2H4', smiles='C=C'), ARCSpecies(label='HCl', smiles='Cl')] + p_species = [ARCSpecies(label='C2H5Cl', smiles='CCCl')] + reaction_dict = ARCReaction(r_species=r_species, p_species=p_species).as_dict() + reaction_dict['family'] = 'XY_Addition_MultipleBond' + with mock.patch.dict(settings, {'rmg_family_set': 'default'}): + rxn = ARCReaction(reaction_dict=reaction_dict) + self.assertEqual(rxn.family, 'XY_Addition_MultipleBond') + self.assertEqual({product_dict['family'] for product_dict in rxn.product_dicts}, + {'XY_Addition_MultipleBond'}) + + def test_a_pinned_arc_family_is_the_family_set_that_is_loaded(self): + """ARC's own families are not RMG families and no recommended family set lists them, so + get_all_families() always appends them and a reaction reaches them unpinned. Pinning one + narrows the loaded family set to it and it still generates its products.""" + def build_rxn(**kwargs): + return ARCReaction(r_species=[ARCSpecies(label='DME', smiles='COC'), + ARCSpecies(label='H2O', smiles='O')], + p_species=[ARCSpecies(label='MeOH_a', smiles='CO'), + ARCSpecies(label='MeOH_b', smiles='CO')], + **kwargs) + loaded_sets = list() + + def spy(**kwargs): + loaded_sets.append(kwargs.get('rmg_family_set')) + return get_reaction_family_products(**kwargs) + with mock.patch.dict(settings, {'rmg_family_set': 'default'}): + self.assertNotIn('ether_hydrolysis', get_rmg_recommended_family_sets()['default']) + self.assertEqual(build_rxn().family, 'ether_hydrolysis') + with mock.patch('arc.reaction.reaction.get_reaction_family_products', side_effect=spy): + product_dicts = build_rxn(family='ether_hydrolysis').product_dicts + self.assertGreater(len(product_dicts), 0) + self.assertEqual({product_dict['family'] for product_dict in product_dicts}, {'ether_hydrolysis'}) + self.assertEqual(loaded_sets, [['ether_hydrolysis']]) + + def test_a_pinned_family_outside_the_configured_set_keeps_own_reverse_and_reverse_discovery(self): + """Cl_Abstraction is its own reverse and lives in the halogens set, so CH3Cl + H <=> CH3 + HCl + has no family under the shipped 'default' set. Pinning it resolves the reaction, keeps the + family's own reverse property, and leaves reverse discovery working.""" + def build_rxn(**kwargs): + return ARCReaction(r_species=[ARCSpecies(label='CH3Cl', smiles='CCl'), + ARCSpecies(label='H', smiles='[H]')], + p_species=[ARCSpecies(label='CH3', smiles='[CH3]'), + ARCSpecies(label='HCl', smiles='Cl')], + **kwargs) + with mock.patch.dict(settings, {'rmg_family_set': 'default'}): + self.assertIsNone(build_rxn().family) + rxn = build_rxn(family='Cl_Abstraction') + self.assertTrue(rxn.family_own_reverse) + self.assertEqual({product_dict['family'] for product_dict in rxn.product_dicts}, {'Cl_Abstraction'}) + self.assertEqual(rxn.determine_family(discover_own_reverse_rxns_in_reverse=True), + ('Cl_Abstraction', True)) + reverse_product_dicts = rxn.get_product_dicts(discover_own_reverse_rxns_in_reverse=True) + self.assertEqual({product_dict['family'] for product_dict in reverse_product_dicts}, {'Cl_Abstraction'}) + + def test_pinning_a_family_on_a_reaction_without_any_family_match_raises(self): + """Ethylperoxy <=> OH + oxirane needs an H to move to the terminal oxygen on top of the ring + closure, which is two family steps, so no available family matches it. Pinning one names the + family in the error rather than yielding no product dicts.""" + rxn = ARCReaction(r_species=[ARCSpecies(label='CH3CH2OO', smiles='CCO[O]')], + p_species=[ARCSpecies(label='OH', smiles='[OH]'), + ARCSpecies(label='oxirane', smiles='C1CO1')], + family='Cyclic_Ether_Formation') + with mock.patch.dict(settings, {'rmg_family_set': 'default'}): + with self.assertRaises(ReactionError) as error: + _ = rxn.product_dicts + self.assertIn('Cyclic_Ether_Formation', str(error.exception)) + self.assertIn('nor any other available reaction family', str(error.exception)) + + def test_pinning_a_family_that_does_not_match_reports_the_families_that_do(self): + """A pinned family narrows what is loaded, so the families the reaction does match are no + longer discovered on the way. The error names them anyway, and names them from the available + families rather than from the configured set: 1,4-cyclohexadiene <=> benzene + H2 belongs to + H2_Loss, which exists only as an RMG database directory and which 'default' does not list.""" + rxn = ARCReaction(r_species=[ARCSpecies(label='1,4-cyclohexadiene', smiles='C1=CCC=CC1')], + p_species=[ARCSpecies(label='benzene', smiles='c1ccccc1'), + ARCSpecies(label='H2', smiles='[H][H]')], + family='H_Abstraction') + with mock.patch.dict(settings, {'rmg_family_set': 'default'}): + self.assertNotIn('H2_Loss', get_all_families()) + with self.assertRaises(ReactionError) as error: + _ = rxn.product_dicts + self.assertIn('H_Abstraction', str(error.exception)) + self.assertIn('H2_Loss', str(error.exception)) + + def test_a_pinned_surface_family_stays_pinnable(self): + """get_all_families('all') skips the surface family sets and directories, so a surface family + is available only through a configured family set that lists it, and pinning one is accepted + under that setting.""" + surface_family = get_rmg_recommended_family_sets()['surface'][0] + with mock.patch.dict(settings, {'rmg_family_set': 'surface'}): + rxn = ARCReaction(r_species=[ARCSpecies(label='C2H4', smiles='C=C')], + p_species=[ARCSpecies(label='C2H4_p', smiles='C=C')], + family=surface_family) + self.assertEqual(rxn.family, surface_family) + rxn.check_family() + + def test_a_lazily_determined_family_does_not_narrow_the_loaded_family_set(self): + """OH + HO2 <=> H2O2 + O matches H_Abstraction and Substitution_O. Reading the family + property determines H_Abstraction, and that determined family is not a pinned family, so a + later generation still loads the configured family set and still finds both.""" + def build_rxn(): + return ARCReaction(r_species=[ARCSpecies(label='OH', smiles='[OH]'), + ARCSpecies(label='HO2', smiles='[O]O')], + p_species=[ARCSpecies(label='H2O2', smiles='OO'), + ARCSpecies(label='O', smiles='[O]')]) + with mock.patch.dict(settings, {'rmg_family_set': 'all'}): + rxn = build_rxn() + self.assertEqual(rxn.family, 'H_Abstraction') + self.assertEqual({product_dict['family'] for product_dict in rxn.get_product_dicts()}, + {'H_Abstraction', 'Substitution_O'}) + self.assertEqual(build_rxn().get_product_dicts(), + build_rxn().get_product_dicts()) + + def test_check_attributes_rejects_an_unavailable_pinned_family(self): + """An unavailable family that reached the reaction through a reaction dictionary is reported + when the reaction is checked, before any product dicts are generated.""" + r_species = [ARCSpecies(label='C2H4', smiles='C=C'), ARCSpecies(label='HCl', smiles='Cl')] + p_species = [ARCSpecies(label='C2H5Cl', smiles='CCCl')] + reaction_dict = ARCReaction(r_species=r_species, p_species=p_species).as_dict() + reaction_dict['family'] = 'Not_A_Family' + rxn = ARCReaction(reaction_dict=reaction_dict) + with self.assertRaises(ReactionError) as error: + rxn.check_attributes() + self.assertIn('Not_A_Family', str(error.exception)) + reaction_dict['family'] = 'XY_Addition_MultipleBond' + with mock.patch.dict(settings, {'rmg_family_set': 'default'}): + ARCReaction(reaction_dict=reaction_dict).check_attributes() + + def test_pinning_an_unavailable_family_raises(self): + """A family name that no RMG or ARC family carries is rejected when it is given as an + argument, and is reported when it reaches the reaction through a reaction dictionary.""" + r_species = [ARCSpecies(label='C2H4', smiles='C=C'), ARCSpecies(label='HCl', smiles='Cl')] + p_species = [ARCSpecies(label='C2H5Cl', smiles='CCCl')] + with self.assertRaises(ValueError): + ARCReaction(r_species=r_species, p_species=p_species, family='Not_A_Family') + reaction_dict = ARCReaction(r_species=r_species, p_species=p_species).as_dict() + reaction_dict['family'] = 'Not_A_Family' + with self.assertRaises(ReactionError) as error: + _ = ARCReaction(reaction_dict=reaction_dict).product_dicts + self.assertIn('Not_A_Family', str(error.exception)) + def test_family_own_reverse_is_derived_from_a_pinned_family(self): """Test that pinning a family also determines whether that family is its own reverse""" rxn = ARCReaction(r_species=[ARCSpecies(label='C2H6', smiles='CC'),