Source code for galaxy.visualization.plugins.registry

"""
Lower level of visualization framework which does three main things:
    - associate visualizations with objects
    - create urls to visualizations based on some target object(s)
    - unpack a query string into the desired objects needed for rendering
"""

import logging
import os
import weakref

import galaxy.model
from galaxy.exceptions import ObjectNotFound
from galaxy.util import config_directories_from_setting
from galaxy.visualization.plugins import (
    config_parser,
    plugin as vis_plugins,
)
from galaxy.visualization.plugins.datasource_testing import is_object_applicable

log = logging.getLogger(__name__)



[docs]
class VisualizationsRegistry:
    """
    Main responsibilities are:
        - discovering visualization plugins in the filesystem
        - testing if an object has a visualization that can be applied to it
        - generating a link to controllers.visualization.render with
            the appropriate params
        - validating and parsing params into resources (based on a context)
            used in the visualization template
    """

    #: base directory of visualizations
    BASE_DIR = "static/plugins/visualizations"
    #: base url to controller endpoint
    BASE_URL = "visualizations"
    #: built-in visualizations
    BUILT_IN_VISUALIZATIONS = ["trackster"]

    def __str__(self):
        return self.__class__.__name__


[docs]
    def __init__(self, app, template_cache_dir=None, directories_setting=None, skip_bad_plugins=True, **kwargs):
        """
        Set up the manager and load all visualization plugins.

        :type   app:        galaxy.app.UniverseApplication
        :param  app:        the application (and its configuration) using this manager
        :type   base_url:   string
        :param  base_url:   url to prefix all plugin urls with
        :type   template_cache_dir: string
        :param  template_cache_dir: filesytem path to the directory where cached
            templates are kept
        """
        self.app = weakref.ref(app)
        self.config_parser = config_parser.VisualizationsConfigParser()
        self.base_url = self.BASE_URL
        self.template_cache_dir = template_cache_dir
        self.skip_bad_plugins = skip_bad_plugins
        self.plugins = {}
        self.directories = config_directories_from_setting(directories_setting, app.config.root)
        self.directories.append(os.path.join(app.config.root, self.BASE_DIR))
        self._load_plugins()


    def _load_plugins(self):
        """
        Search ``self.directories`` for potential plugins, load them, and cache
        in ``self.plugins``.
        """
        for plugin_path in self._find_plugins():
            try:
                plugin = self._load_plugin(plugin_path)
                if plugin and plugin.name not in self.plugins:
                    self.plugins[plugin.name] = plugin
                    log.info("%s, loaded plugin: %s", self, plugin.name)
                elif plugin and plugin.name in self.plugins:
                    log.warning("%s, plugin with name already exists: %s. Skipping...", self, plugin.name)
            except Exception:
                if not self.skip_bad_plugins:
                    raise
                log.exception("Plugin loading raised exception: %s. Skipping...", plugin_path)
        return self.plugins

    def _find_plugins(self):
        """
        Return the directory paths of plugins within ``self.directories``.

        Paths are considered a plugin path if they pass ``self.is_plugin``.
        :rtype:                 string generator
        :returns:               paths of valid plugins
        """
        # due to the ordering of listdir, there is an implicit plugin loading order here
        # could instead explicitly list on/off in master config file
        for directory in self.directories:
            if not os.path.isdir(directory):
                continue
            for plugin_dir in sorted(os.listdir(directory)):
                plugin_path = os.path.join(directory, plugin_dir)
                if self._is_plugin(plugin_path):
                    yield plugin_path
                if os.path.isdir(plugin_path):
                    for plugin_subdir in sorted(os.listdir(plugin_path)):
                        plugin_subpath = os.path.join(plugin_path, plugin_subdir)
                        if self._is_plugin(plugin_subpath):
                            yield plugin_subpath

    def _is_plugin(self, plugin_path):
        """
        Determines whether the given filesystem path contains a plugin.

        In this base class, all sub-directories are considered plugins.

        :type   plugin_path:    string
        :param  plugin_path:    relative or absolute filesystem path to the
            potential plugin
        :rtype:                 bool
        :returns:               True if the path contains a plugin
        """
        # plugin_path must be a directory, have a config dir, and a config file matching the plugin dir name
        if not os.path.isdir(plugin_path):
            # super won't work here - different criteria
            return False
        expected_config_filename = f"{os.path.basename(plugin_path)}.xml"
        return os.path.isfile(os.path.join(plugin_path, "static", expected_config_filename))

    def _load_plugin(self, plugin_path):
        """
        Create the visualization plugin object, parse its configuration file,
        and return it.

        :type   plugin_path:    string
        :param  plugin_path:    relative or absolute filesystem path to the plugin
        :rtype:                 ``VisualizationPlugin``
        :returns:               the loaded plugin
        """
        plugin_name = os.path.split(plugin_path)[1]
        config_file = os.path.join(plugin_path, "static", f"{plugin_name}.xml")
        if os.path.exists(config_file):
            config = self.config_parser.parse_file(config_file)
            if config is not None:
                plugin = self._build_plugin(plugin_name, plugin_path, config)
                return plugin
        raise ObjectNotFound(f"Visualization XML not found in config or static paths for: {plugin_name}.")

    def _build_plugin(self, plugin_name, plugin_path, config):
        # from mako
        if config["entry_point"]["type"] == "mako":
            plugin_class = vis_plugins.VisualizationPlugin
        else:
            # default from js
            plugin_class = vis_plugins.ScriptVisualizationPlugin
        return plugin_class(
            self.app(),
            plugin_path,
            plugin_name,
            config,
            context=dict(
                base_url=self.base_url,
                template_cache_dir=self.template_cache_dir,
            ),
        )


[docs]
    def get_plugin(self, key):
        """
        Wrap to throw error if plugin not in registry.
        """
        if key not in self.plugins:
            raise ObjectNotFound(f"Unknown or invalid visualization: {key}")
        return self.plugins[key]


    # -- building links to visualizations from objects --

[docs]
    def get_visualizations(self, trans, target_object=None, embeddable=None):
        """
        Get the names of visualizations usable on the `target_object` and
        the urls to call in order to render the visualizations.
        """
        result = []
        for vis_name, vis_plugin in self.plugins.items():
            if vis_plugin.config.get("hidden"):
                continue
            if embeddable and not vis_plugin.config.get("embeddable"):
                continue
            if target_object is not None and self.get_visualization(trans, vis_name, target_object) is None:
                continue
            result.append(vis_plugin.to_dict())
        return sorted(result, key=lambda k: k.get("html"))



[docs]
    def get_visualization(self, trans, visualization_name, target_object):
        """
        Return data to build a url to the visualization with the given
        `visualization_name` if it's applicable to `target_object` or
        `None` if it's not.
        """
        if (visualization := self.plugins.get(visualization_name, None)) is not None:
            data_sources = visualization.config["data_sources"]
            for data_source in data_sources:
                model_class = data_source["model_class"]
                model_class = getattr(galaxy.model, model_class, None)
                if model_class and isinstance(target_object, model_class):
                    tests = data_source["tests"]
                    if tests is None or is_object_applicable(trans, target_object, tests):
                        return visualization.to_dict()