Let a pinned reaction family determine which families are loaded

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

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"""

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,
@@ -111,10 +115,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}")
@@ -227,12 +232,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
@@ -556,23 +563,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,
@@ -596,10 +634,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):
@@ -641,37 +676,54 @@ 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.
         """
         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: