rmg_sa.py replaces run_sa function in main.py

Also remove all of the corresponding imports in main.py that have now been moved to the RMG SA Adapter: rmg_sa.py

tandem/main.py

@@ -88,12 +88,8 @@
 from rmgpy.rmg.main import initialize_log as initialize_rmg_log
 from rmgpy.rmg.main import RMG
 from rmgpy.rmg.pdep import PDepReaction
-from rmgpy.solver.simple import SimpleReactor
-from rmgpy.solver.liquid import LiquidReactor
 from rmgpy.species import Species
 from rmgpy.thermo import NASAPolynomial, NASA, ThermoData, Wilhoit
-from rmgpy.tools.loader import load_rmg_py_job
-from rmgpy.tools.simulate import simulate
 
 from arc.common import get_ordinal_indicator, key_by_val, read_yaml_file, time_lapse
 from arc.main import ARC
@@ -107,6 +103,7 @@
      log_species_to_calculate,
      log_unconverged_species,
      )
+from .sa.factory import sa_factory
 from .utils import combine_dicts, delete_root_rmg_log, dict_to_str
 
 if TYPE_CHECKING:
@@ -188,12 +185,16 @@ def execute(args,
             run_rmg(input_file=rmg_input_file, output_directory=run_directory, kwargs=kwargs, arguments=arguments,
                     tolerance=tolerance, thermo_library=thermo_library)
 
-        sa_success = run_sa(method=arguments['SA method'],
-                            observable_list=arguments['SA observables'],
-                            input_file=rmg_input_file,
-                            run_directory=run_directory,
-                            threshold=arguments['SA threshold'],
-                            )
+        sa_adapter = sa_factory(sa_method=arguments['SA method'],
+                                observable_list=arguments['SA observables'],
+                                run_directory=run_directory,
+                                input_file=rmg_input_file,
+                                threshold=arguments['SA threshold'],
+                                verbose=verbose,
+                                )
+
+        sa_success = sa_adapter.run_sa()
+
         if not sa_success:
             log(f'Could not complete the sensitivity analysis using {arguments["SA method"]}', 'error', verbose=verbose)
             if len(tolerances) > i + 1:
@@ -371,106 +372,6 @@ def run_arc(input_dict: Union[str, dict],
     log(f'ARC terminated. Overall execution time: {elapsed_time}', verbose=verbose)
 
 
-def run_sa(method: str,
-           observable_list: list,
-           run_directory: str,
-           input_file: str,
-           threshold: Optional[float] = 0.001,
-           verbose: Optional[bool] = True,
-           ) -> bool:
-    """
-    Run a sensitivity analysis.
-
-    Args:
-        method (str): The software to use, either RMG, RMS, or Cantera.
-        observable_list (list): Entries are dictionaries of 'label' and structure (either 'smiles' or 'adj').
-        run_directory (str): A path to the RMG-ARC iteration directory.
-        input_file (str): The path to the legacy RMG input file.
-        threshold (Optional[float]): The sensitivity threshold to use.
-        verbose (Optional[bool]): Whether or not to log to file.
-
-    Raises:
-        InputError: If ``method`` has a wrong value.
-
-    Returns:
-        bool: Whether SA ran successfully. ``True`` if it did.
-    """
-    if method.lower() not in ['rmg', 'rms', 'cantera']:
-        raise InputError(f'The "SA method" argument must equal to either "RMG", "RMS", or "Cantera". Got: {method}')
-    if method.lower() == 'rmg':
-        log('Running SA using RMG...', verbose=verbose)
-        model = os.path.join(run_directory, 'chemkin', 'chem_annotated.inp')
-        species_dict = os.path.join(run_directory, 'chemkin', 'species_dictionary.txt')
-        sa_path = os.path.join(run_directory, 'sa')
-        if not os.path.isdir(sa_path):
-            os.mkdir(sa_path)
-        rmg_sa_input_file = os.path.join(sa_path, 'input.py')
-        if os.path.isfile(rmg_sa_input_file):
-            os.remove(rmg_sa_input_file)
-        shutil.copyfile(src=input_file, dst=rmg_sa_input_file)
-
-        rmg = load_rmg_py_job(input_file=rmg_sa_input_file,
-                              chemkin_file=model,
-                              species_dict=species_dict,
-                              generate_images=True,
-                              use_chemkin_names=False,
-                              check_duplicates=False)
-
-        rmg_species = rmg.reaction_model.core.species
-        rmg_observable_species = list()
-        for observable in observable_list:
-            for rmg_spc in rmg_species:
-                if 'adj' in observable:
-                    observable_spc = Species(label=observable['label']).from_adjacency_list(observable['adj'])
-                elif 'smiles' in observable:
-                    observable_spc = Species(label=observable['label']).from_smiles(observable['smiles'])
-                else:
-                    raise InputError(f'All SA observables must have structure (smiles or adj), got: {observable}')
-                if observable_spc.label == rmg_spc.label or observable_spc.is_isomorphic(rmg_spc):
-                    rmg_observable_species.append(rmg_spc)
-                    break
-            else:
-                raise InputError(f'Could not find the observable species {observable["label"]} '
-                                 f'in the RMG species list.')
-
-        for reaction_system in rmg.reaction_systems:
-            if isinstance(reaction_system, SimpleReactor):
-                reaction_system.sensitive_species = rmg_observable_species
-                reaction_system.sensitivity_threshold = threshold
-                if hasattr(reaction_system, 'Trange') and reaction_system.Trange is not None:
-                    temperature = sum([t.value_si for t in reaction_system.Trange]) / len(reaction_system.Trange)
-                else:
-                    temperature = reaction_system.T.value_si
-                reaction_system.sens_conditions['T'] = temperature
-                if hasattr(reaction_system, 'Prange') and reaction_system.Prange is not None:
-                    pressure = sum([p.value_si for p in reaction_system.Prange]) / len(reaction_system.Prange)
-                else:
-                    pressure = reaction_system.P.value_si
-                reaction_system.sens_conditions['P'] = pressure
-            elif isinstance(reaction_system, LiquidReactor):
-                reaction_system.sensitive_species = rmg_observable_species
-                reaction_system.sensitivity_threshold = threshold
-                if hasattr(reaction_system, 'Trange') and reaction_system.Trange is not None:
-                    temperature = sum([t.value_si for t in reaction_system.Trange]) / len(reaction_system.Trange)
-                else:
-                    temperature = reaction_system.T.value_si
-                reaction_system.sens_conditions['T'] = temperature
-                if hasattr(reaction_system, 'Vrange') and reaction_system.Vrange is not None:
-                    volume = sum([v for v in reaction_system.Vrange]) / len(reaction_system.Vrange)
-                else:
-                    volume = reaction_system.V
-                reaction_system.sens_conditions['V'] = volume
-            else:
-                raise NotImplementedError(f'RMG SA not implemented for Reactor type {type(reaction_system)}.')
-        try:
-            simulate(rmg)
-        except FileNotFoundError:
-            return False
-    else:
-        raise NotImplementedError('Currently only RMG is implemented as an SA method')  # temp
-    return True
-
-
 def restart_t3(path: str,
                thermo_library: Optional[str] = None,
                verbose: Optional[bool] = True,