"""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 typing import (
Any,
Dict,
Optional,
TYPE_CHECKING,
)
from urllib.request import urlopen
from planemo.config import read_global_config
if TYPE_CHECKING:
from planemo.config import OptionSource
[docs]
class PlanemoContextInterface(metaclass=abc.ABCMeta):
"""Interface Planemo operations use to access workspace context."""
[docs]
@abc.abstractmethod
def set_option_source(self, param_name, option_source, force=False):
"""Specify how an option was set."""
[docs]
@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."""
[docs]
@abc.abstractmethod
def log(self, msg, *args):
"""Log a message."""
[docs]
@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."""
[docs]
@abc.abstractmethod
def cache_download(self, url, destination):
"""Use workspace to cache download of ``url``."""
[docs]
class PlanemoContext(PlanemoContextInterface):
"""Implementation of ``PlanemoContextInterface``"""
planemo_directory: Optional[str]
def __init__(self) -> None:
"""Construct a Context object using execution environment."""
self.home = os.getcwd()
self._global_config: Optional[Dict] = None
# Will be set by planemo CLI driver
self.verbose = False
self.planemo_config: Optional[str] = None
self.planemo_directory = None
self.option_source: Dict[str, "OptionSource"] = {}
[docs]
def set_option_source(self, param_name: str, option_source: "OptionSource", force: bool = False) -> None:
"""Specify how an option was set."""
if not force:
assert param_name not in self.option_source, f"Option source for [{param_name}] already set"
self.option_source[param_name] = option_source
[docs]
def get_option_source(self, param_name: str) -> "OptionSource":
"""Return OptionSource value indicating how the option was set."""
assert param_name in self.option_source, f"No option source for [{param_name}]"
return self.option_source[param_name]
@property
def global_config(self) -> Dict[str, Any]:
"""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 {}
[docs]
def log(self, msg: str, *args) -> None:
"""Log a message."""
if args:
msg %= args
self._log_message(msg)
[docs]
def vlog(self, msg: str, *args, **kwds) -> None:
"""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) -> str:
"""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) -> str:
"""Create a return a directory for storing Galaxy profiles."""
path = os.path.join(self.workspace, "profiles")
return self._ensure_directory(path, "Galaxy profiles")
[docs]
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(f"Failed to download [{url}].")
with open(cache_destination, "wb") as f:
f.write(content)
shutil.copy(cache_destination, destination)
def _ensure_directory(self, path: str, name: str) -> str:
if not os.path.exists(path):
os.makedirs(path)
if not os.path.isdir(path):
raise Exception(f"Planemo {name} directory [{path}] unavailable.")
return path
def _log_message(self, message):
"""Extension point for overriding I/O."""
print(message)
__all__ = (
"configure_standard_planemo_logging",
"PlanemoContextInterface",
"PlanemoContext",
)