diff env/lib/python3.9/site-packages/planemo/context.py @ 0:4f3585e2f14b draft default tip

"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/env/lib/python3.9/site-packages/planemo/context.py	Mon Mar 22 18:12:50 2021 +0000
@@ -0,0 +1,213 @@
+"""Define the context around Planemo computation.
+
+Abstractions for cross cutting concerns (logging, workspace management,
+etc.).
+"""
+import abc
+import logging.config
+import os
+import shutil
+import sys
+import traceback
+from urllib.request import urlopen
+
+from planemo.config import read_global_config
+
+
+class PlanemoContextInterface(metaclass=abc.ABCMeta):
+    """Interface Planemo operations use to access workspace context."""
+
+    @abc.abstractmethod
+    def set_option_source(self, param_name, option_source, force=False):
+        """Specify how an option was set."""
+
+    @abc.abstractmethod
+    def get_option_source(self, param_name):
+        """Return OptionSource value indicating how the option was set."""
+
+    @abc.abstractproperty
+    def global_config(self):
+        """Read Planemo's global configuration."""
+
+    @abc.abstractmethod
+    def log(self, msg, *args):
+        """Log a message."""
+
+    @abc.abstractmethod
+    def vlog(self, msg, *args, **kwds):
+        """Log a message only if verbose is enabled."""
+
+    @abc.abstractproperty
+    def workspace(self):
+        """Create and return Planemo's workspace."""
+
+    @abc.abstractproperty
+    def galaxy_profiles_directory(self):
+        """Create a return a directory for storing Galaxy profiles."""
+
+    @abc.abstractmethod
+    def cache_download(self, url, destination):
+        """Use workspace to cache download of ``url``."""
+
+
+class PlanemoContext(PlanemoContextInterface):
+    """Implementation of ``PlanemoContextInterface``"""
+
+    def __init__(self):
+        """Construct a Context object using execution environment."""
+        self.home = os.getcwd()
+        self._global_config = None
+        # Will be set by planemo CLI driver
+        self.verbose = False
+        self.planemo_config = None
+        self.planemo_directory = None
+        self.option_source = {}
+
+    def set_option_source(self, param_name, option_source, force=False):
+        """Specify how an option was set."""
+        if not force:
+            assert param_name not in self.option_source, "No option source for [%s]" % param_name
+        self.option_source[param_name] = option_source
+
+    def get_option_source(self, param_name):
+        """Return OptionSource value indicating how the option was set."""
+        assert param_name in self.option_source, "No option source for [%s]" % param_name
+        return self.option_source[param_name]
+
+    @property
+    def global_config(self):
+        """Read Planemo's global configuration.
+
+        As defined most simply by ~/.planemo.yml.
+        """
+        if self._global_config is None:
+            self._global_config = read_global_config(self.planemo_config)
+        return self._global_config or {}
+
+    def log(self, msg, *args):
+        """Log a message."""
+        if args:
+            msg %= args
+        self._log_message(msg)
+
+    def vlog(self, msg, *args, **kwds):
+        """Log a message only if verbose is enabled."""
+        if self.verbose:
+            self.log(msg, *args)
+            if kwds.get("exception", False):
+                traceback.print_exc(file=sys.stderr)
+
+    @property
+    def workspace(self):
+        """Create and return Planemo's workspace.
+
+        By default this will be ``~/.planemo``.
+        """
+        if not self.planemo_directory:
+            raise Exception("No planemo workspace defined.")
+        workspace = self.planemo_directory
+        return self._ensure_directory(workspace, "workspace")
+
+    @property
+    def galaxy_profiles_directory(self):
+        """Create a return a directory for storing Galaxy profiles."""
+        path = os.path.join(self.workspace, "profiles")
+        return self._ensure_directory(path, "Galaxy profiles")
+
+    def cache_download(self, url, destination):
+        """Use workspace to cache download of ``url``."""
+        cache = os.path.join(self.workspace, "cache")
+        if not os.path.exists(cache):
+            os.makedirs(cache)
+        filename = os.path.basename(url)
+        cache_destination = os.path.join(cache, filename)
+        if not os.path.exists(cache_destination):
+            with urlopen(url) as fh:
+                content = fh.read()
+            if len(content) == 0:
+                raise Exception("Failed to download [%s]." % url)
+            with open(cache_destination, "wb") as f:
+                f.write(content)
+
+        shutil.copy(cache_destination, destination)
+
+    def _ensure_directory(self, path, name):
+        if not os.path.exists(path):
+            os.makedirs(path)
+        if not os.path.isdir(path):
+            template = "Planemo %s directory [%s] unavailable."
+            message = template % (name, path)
+            raise Exception(message)
+        return path
+
+    def _log_message(self, message):
+        """Extension point for overriding I/O."""
+        print(message)
+
+
+def configure_standard_planemo_logging(verbose):
+    """Configure Planemo's default logging rules."""
+    logging_config = {
+        'version': 1,
+        'disable_existing_loggers': False,
+        'formatters': {
+            'verbose': {
+                'format': '%(name)s %(levelname)s %(asctime)s: %(message)s'
+            },
+            'simple': {
+                'format': '%(name)s %(levelname)s: %(message)s'
+            },
+        },
+        'handlers': {
+            'console': {
+                'level': 'DEBUG',
+                'class': 'logging.StreamHandler',
+                'formatter': 'simple' if not verbose else 'verbose'
+            },
+        },
+        'loggers': {
+            # Suppress CWL is beta warning, for Planemo purposes - it is absolutely not.
+            'galaxy.tools.parser.factory': {
+                'handlers': ['console'],
+                'propagate': False,
+                'level': 'ERROR' if not verbose else "DEBUG",
+            },
+            'galaxy.tools.deps.commands': {
+                'handlers': ['console'],
+                'propagate': False,
+                'level': 'ERROR' if not verbose else "DEBUG",
+            },
+            'galaxy': {
+                'handlers': ['console'],
+                'propagate': False,
+                'level': 'INFO' if not verbose else "DEBUG",
+            },
+            # @jmchilton
+            # I'm fixing up Planemo's lint functionality for CWL and I keep seeing this for the
+            # schema metadata stuff (e.g. in the workflows repo). "rdflib.term WARNING:
+            # http://schema.org/docs/!DOCTYPE html does not look like a valid URI, trying to
+            # serialize this will break.". I'm going to suppress this warning I think, or are the
+            # examples wrong and should declare their namespaces differently in some way?
+            # @mr-c
+            # That particular warning is worth suppressing. A PR to silence it permanently would be very welcome!
+            # https://github.com/RDFLib/rdflib/blob/master/rdflib/term.py#L225
+            'rdflib.term': {
+                'handlers': ['console'],
+                'propagate': False,
+                'level': 'ERROR' if not verbose else "DEBUG",
+            }
+        },
+        'root': {
+            'handlers': ['console'],
+            'propagate': False,
+            'level': 'WARNING' if not verbose else "DEBUG",
+        }
+    }
+    logging.config.dictConfig(logging_config)
+
+
+__all__ = (
+    'configure_standard_planemo_logging',
+    'PlanemoContextInterface',
+    'PlanemoContext',
+)