Add AnalysisScript base class and esnb derived class to define run_analysis for esnb analysis scripts

Author: Chris Blanton

fre/analysis/base_class.py

@@ -0,0 +1,42 @@
+import json
+
+
+class AnalysisScript(object):
+    """Abstract base class for analysis scripts.  User-defined analysis scripts
+       should inhert from this class and override the requires and run_analysis methods.
+
+    Attributes:
+       description: Longer form description for the analysis.
+       title: Title that describes the analysis.
+    """
+    def __init__(self):
+        """Instantiates an object.  The user should provide a description and title."""
+        raise NotImplementedError("you must override this function.")
+        self.description = None
+        self.title = None
+
+    def requires(self):
+        """Provides metadata describing what is needed for this analysis to run.
+
+        Returns:
+            A json string describing the metadata.
+        """
+        raise NotImplementedError("you must override this function.")
+        return json.dumps("{json of metadata MDTF format.}")
+
+    def run_analysis(self, yaml, name, date_range, scripts_dir, output_dir, output_yaml):
+        """Runs the analysis and generates all plots and associated datasets.
+
+        Args:
+            yaml: Path to a model yaml
+            name: Name of the analysis as specified in the yaml
+            date_range: Time span to use for analysis (YYYY-MM-DD,YYYY-MM-DD)
+            scripts_dir: Path to a directory to save intermediate scripts
+            output_dir: Path to a directory to save figures
+            output_yaml: Path to use as an structured output yaml file
+
+        Returns:
+            A list of png figures.
+        """
+        raise NotImplementedError("you must override this function.")
+        return ["figure1.png", "figure2.png",]

fre/analysis/env_tool.py

@@ -0,0 +1,132 @@
+from pathlib import Path
+from subprocess import CalledProcessError, PIPE, run, STDOUT
+from tempfile import TemporaryDirectory
+import venv
+
+
+def _process_output(output):
+    """Converts bytes string to list of String lines.
+
+    Args:
+        output: Bytes string.
+
+    Returns:
+        List of strings.
+    """
+    return [x for x in output.decode("utf-8").split("\n") if x]
+
+
+class VirtualEnvManager(object):
+    """Helper class for creating/running simple command in a virtual environment."""
+    def __init__(self, path):
+        self.path = Path(path)
+        self.activate = f"source {self.path / 'bin' / 'activate'}"
+
+    @staticmethod
+    def _execute(commands):
+        """Runs input commands through bash in a child process.
+
+        Args:
+            commands: List of string commands.
+
+        Returns:
+            List of string output.
+        """
+        with TemporaryDirectory() as tmp:
+            script_path = Path(tmp) / "script"
+            with open(script_path, "w") as script:
+                script.write("\n".join(commands))
+            try:
+                process = run(["bash", str(script_path)], stdout=PIPE, stderr=STDOUT,
+                              check=True)
+            except CalledProcessError as err:
+                for line in _process_output(err.output):
+                    print(line)
+                raise
+            return _process_output(process.stdout)
+
+    def _execute_python_script(self, commands):
+        """Runs input python code in bash in a child process.
+
+        Args:
+            commands: List of string python code lines.
+
+        Returns:
+            List of string output.
+        """
+        with TemporaryDirectory() as tmp:
+            script_path = Path(tmp) / "python_script"
+            with open(script_path, "w") as script:
+                script.write("\n".join(commands))
+            commands = [self.activate, f"python3 {str(script_path)}"]
+            return self._execute(commands)
+
+    def create_env(self):
+        """Creates the virtual environment."""
+        venv.create(self.path, with_pip=True)
+
+    def destroy_env(self):
+        """Destroys the virtual environment."""
+        raise NotImplementedError("this feature is not implemented yet.")
+
+    def install_package(self, name):
+        """Installs a package in the virtual environment.
+
+        Args:
+            name: String name of the package.
+
+        Returns:
+            List of string output.
+        """
+        commands = [self.activate, "python3 -m pip --upgrade pip",
+                    f"python3 -m pip install {name}"]
+        return self._execute(commands)
+
+    def list_plugins(self):
+        """Returns a list of plugins that are available in the virtual environment.
+
+        Returns:
+            List of plugins.
+        """
+        python_script = [
+            "from analysis_scripts import available_plugins",
+            "for plugin in available_plugins():",
+            "    print(plugin)"
+        ]
+        return self._execute_python_script(python_script)
+
+    def run_analysis_plugin(self, name, catalog, output_directory, config=None):
+        """Returns a list of paths to figures created by the plugin from the virtual
+           environment.
+
+        Args:
+             name: String name of the analysis package.
+             catalog: Path to the data catalog.
+             output_directory: Path to the output directory.
+
+        Returns:
+            List of figure paths.
+        """
+        if config:
+            python_script = [f"config = {str(config)}",]
+        else:
+            python_script = ["config = None",]
+        python_script += [
+            "from analysis_scripts import run_plugin",
+            f"paths = run_plugin('{name}', '{catalog}', '{output_directory}', config=config)",
+            "for path in paths:",
+            "    print(path)"
+        ]
+        return self._execute_python_script(python_script)
+
+    def uninstall_package(self, name):
+        """Uninstalls a package from the virtual environment.
+
+        Args:
+            name: String name of the package.
+
+        Returns:
+            List of string output.
+        """
+        commands = [self.activate, f"pip uninstall {name}"]
+        return self._execute(commands)

fre/analysis/plugins/__init__.py

@@ -0,0 +1 @@
+from .esnb import freanalysis_esnb

fre/analysis/plugins/esnb.py

@@ -0,0 +1,57 @@
+import logging
+from pathlib import Path, PurePosixPath
+import requests
+from ..base_class import AnalysisScript
+import esnb.engine
+
+fre_logger = logging.getLogger(__name__)
+
+class freanalysis_esnb(AnalysisScript):
+    """Defines run and report-requirements methods for ESNB flavor usage
+    """
+
+    def __init__(self):
+        self.description = "Wrapper to access analysis framework for ESNB scripts"
+        self.title = "ESNB"
+
+    def run_analysis(self, config, name, date_range, scripts_dir, output_dir, output_yaml):
+        """Runs the ESNB analysis specified in the yaml and the runtime options
+
+        Args:
+            config: Dictionary of specific configuration for the script
+            name: Name of the analysis as specified in the yaml
+            date_range: Time span to use for analysis (YYYY-MM-DD,YYYY-MM-DD)
+            scripts_dir: Path to a directory to save intermediate scripts
+            output_dir: Path to a directory to save figures
+            output_yaml: Path to use as an structured output yaml file
+        """
+
+        # save notebook to scripts_dir
+        url = config["notebook_path"]
+        # convert to the "Raw" URL
+        # replace 'github.com' with 'raw.githubusercontent.com' and remove '/blob'
+        raw_url = url.replace("github.com", "raw.githubusercontent.com").replace("/blob/", "/")
+        local_filename = Path(scripts_dir) / PurePosixPath(url).name
+        with requests.get(raw_url) as r:
+            r.raise_for_status() # Check for HTTP errors (404, 500, etc.)
+            with open(local_filename, 'wb') as f:
+                for chunk in r.iter_content(chunk_size=8192):
+                    f.write(chunk)
+        fre_logger.debug(f"ESNB notebook saved to '{local_filename}'")
+
+        # create run_settings dictionary
+        run_settings = {
+            'conda_env_root': config["conda_env_root"],
+            'notebook_path': local_filename,
+            'outdir': output_dir,
+            'scripts_dir': scripts_dir
+        }
+
+        # create case_settings dictionary
+
+        # write the python script that runs the notebook
+        python_script = esnb.engine.canopy_launcher(run_settings, verbose=True) 
+        fre_logger.debug(f"ESNB python wrapper saved to '{python_script}'")
+
+        # run the python script
+

fre/analysis/plugins/subtools.py

@@ -0,0 +1,143 @@
+import importlib
+import inspect
+from pathlib import Path
+import pkgutil
+
+from ..base_class import AnalysisScript
+from .esnb import freanalysis_esnb
+
+
+class UnknownPluginError(BaseException):
+    """Custom exception for when an invalid plugin name is used."""
+    pass
+
+
+def _find_plugin_class(module):
+    """Looks for a class that inherits from AnalysisScript.
+
+    Args:
+        module: Module object.
+
+    Returns:
+        Class that inherits from AnalysisScript.
+
+    Raises:
+        UnknownPluginError if no class is found.
+    """
+    for attribute in vars(module).values():
+        # Try to find a class that inherits from the AnalysisScript class.
+        if inspect.isclass(attribute) and AnalysisScript in attribute.__bases__:
+            # Return the class so an object can be instantiated from it later.
+            return attribute
+    raise UnknownPluginError("could not find class that inherts from AnalysisScripts")
+
+
+_sanity_counter = 0  # How much recursion is happening.
+_maximum_craziness = 100  # This is too much recursion.
+
+
+def _recursive_search(name, ispkg):
+    """Recursively search for a module that has a class that inherits from AnalysisScript.
+
+    Args:
+        name: String name of the module.
+        ispkg: Flag telling whether or not the module is a package.
+
+    Returns:
+        Class that inherits from AnalysisScript.
+
+    Raises:
+        UnknownPluginError if no class is found.
+        ValueError if there is too much recursion.
+    """
+    global _sanity_counter
+    _sanity_counter += 1
+    if _sanity_counter > _maximum_craziness:
+        raise ValueError(f"recursion level {_sanity_counter} too high.")
+
+    module = importlib.import_module(name)
+    try:
+        return _find_plugin_class(module)
+    except UnknownPluginError:
+        if not ispkg:
+            # Do not recurse further.
+            raise
+        paths = module.__spec__.submodule_search_locations
+        for finder, subname, ispkg in pkgutil.iter_modules(paths):
+            subname = f"{name}.{subname}"
+            try:
+                return _recursive_search(subname, ispkg)
+            except UnknownPluginError:
+                # Didn't find it, so continue to iterate.
+                pass
+
+
+# Dictionary of found plugins.
+_discovered_plugins = {}
+for finder, name, ispkg in pkgutil.iter_modules():
+    if name.startswith("freanalysis_") and ispkg:
+        _sanity_counter = 0
+        _discovered_plugins[name] = _recursive_search(name, True)
+
+
+def _plugin_object(name):
+    """Attempts to create an object from a class that inherits from AnalysisScript in
+       the plugin module.
+
+    Args:
+        name: Name of the plugin.
+
+    Returns:
+        The object that inherits from AnalysisScript.
+
+    Raises:
+        UnknownPluginError if the input name is not in the disovered_plugins dictionary.
+    """
+    return freanalysis_esnb()
+#    try:
+        #return _discovered_plugins[name]()
+#        return freanalysis_esnb()
+#    except KeyError:
+#        raise UnknownPluginError(f"could not find analysis script plugin '{name}'.")
+
+
+def available_plugins():
+    """Returns a list of plugin names."""
+    return sorted(list(_discovered_plugins.keys()))
+
+
+def list_plugins():
+    """Prints a list of plugin names."""
+    names = available_plugins()
+    if names:
+        print("\n".join(["Available plugins:", "-"*32] + names))
+    else:
+        print("Warning: no plugins found.")
+
+
+def plugin_requirements(name):
+    """Returns a JSON string detailing the plugin's requirement metadata.
+
+    Args:
+        name: Name of the plugin.
+
+    Returns:
+        JSON string of metadata.
+    """
+    return _plugin_object(name).requires()
+
+
+def run_plugin(script_type, name, config, date_range, scripts_dir, output_dir, output_yaml):
+    """Runs the plugin's analysis.
+
+    Args:
+        name: Name of the plugin.
+        catalog: Path to the data catalog.
+        png_dir: Directory where the output figures will be stored.
+        config: Dictionary of configuration values.
+        catalog: Path to the catalog of reference data.
+
+    Returns:
+        A list of png figure files that were created by the analysis.
+    """
+    return _plugin_object(script_type).run_analysis(config, name, date_range, scripts_dir, output_dir, output_yaml)