Source code for galaxy.tool_util.deps.resolvers.modules

This file contains the outline of an implementation to load environment modules

This is a community contributed feature and the core Galaxy team does utilize
it, hence support for it will be minimal. The Galaxy team eagerly welcomes
community contribution and maintenance however.
import logging
from io import StringIO
from os import (
from os.path import (
from subprocess import (

from . import (

log = logging.getLogger(__name__)

DEFAULT_MODULECMD_PATH = "modulecmd"  # Just check path
DEFAULT_MODULE_PATH = '/usr/share/modules/modulefiles'
DEFAULT_MAPPING_FILE = 'config/environment_modules_mapping.yml'
UNKNOWN_FIND_BY_MESSAGE = "ModuleDependencyResolver does not know how to find modules by [%s], find_by should be one of %s"

[docs]class ModuleDependencyResolver(DependencyResolver, MappableDependencyResolver): dict_collection_visible_keys = DependencyResolver.dict_collection_visible_keys + ['base_path', 'modulepath', 'modulecmd', 'prefetch', 'default_indicator', 'find_by'] resolver_type = "modules"
[docs] def __init__(self, dependency_manager, **kwds): # Mapping file management self._set_default_mapping_file(kwds) self._setup_mapping(dependency_manager, **kwds) self.versionless = _string_as_bool(kwds.get('versionless', 'false')) find_by = kwds.get('find_by', 'avail') self.find_by = find_by prefetch = _string_as_bool(kwds.get('prefetch', DEFAULT_MODULE_PREFETCH)) self.modulecmd = kwds.get('modulecmd', DEFAULT_MODULECMD_PATH) self.modulepath = kwds.get('modulepath', self.__default_modulespath()) self.default_indicator = kwds.get('default_indicator', DEFAULT_INDICATOR) if find_by == 'directory': self.module_checker = DirectoryModuleChecker(self, self.modulepath, prefetch) elif find_by == 'avail': self.module_checker = AvailModuleChecker(self, self.modulepath, prefetch, self.default_indicator) else: raise Exception(UNKNOWN_FIND_BY_MESSAGE % (find_by, ["avail", "directory"]))
def __default_modulespath(self): if 'MODULEPATH' in environ: module_path = environ['MODULEPATH'] elif 'MODULESHOME' in environ: module_path = join(environ['MODULESHOME'], 'modulefiles') else: module_path = DEFAULT_MODULE_PATH return module_path def _set_default_mapping_file(self, resolver_attributes): if 'mapping_files' not in resolver_attributes: if exists(DEFAULT_MAPPING_FILE): resolver_attributes['mapping_files'] = DEFAULT_MAPPING_FILE
[docs] def resolve(self, requirement, **kwds): requirement = self._expand_mappings(requirement) name, version, type =, requirement.version, requirement.type if type != "package": return NullDependency(version=version, name=name) if self.__has_module(name, version): return ModuleDependency(self, name, version, exact=True, dependency_resolver=self) elif self.versionless and self.__has_module(name, None): return ModuleDependency(self, name, None, exact=False, dependency_resolver=self) return NullDependency(version=version, name=name)
def __has_module(self, name, version): return self.module_checker.has_module(name, version)
class DirectoryModuleChecker: """Finds module by path. Searches the paths listed in modulepath to for a file or directory matching the module name. If the version=True, searches for files named module/version.""" def __init__(self, module_dependency_resolver, modulepath, prefetch): self.module_dependency_resolver = module_dependency_resolver self.directories = modulepath.split(pathsep) if prefetch: log.warning("Created module dependency resolver with prefetch enabled, but directory module checker does not support this.") def has_module(self, module, version): has_module = False for directory in self.directories: module_directory = join(directory, module) has_module_directory = isdir(module_directory) if not version: has_module = has_module_directory or exists(module_directory) # could be a bare modulefile else: modulefile = join(module_directory, version) has_modulefile = exists(modulefile) has_module = has_module_directory and has_modulefile if has_module: break return has_module class AvailModuleChecker: """Finds modules by searching output of 'module avail'. Parses the Environment Modules 'module avail' output, splitting module names into module and version on '/' and discarding a postfix matching default_indicator (by default '(default)'. Matching is done using the module and (if version=True) the module version.""" def __init__(self, module_dependency_resolver, modulepath, prefetch, default_indicator=DEFAULT_INDICATOR): self.module_dependency_resolver = module_dependency_resolver self.modulepath = modulepath self.default_indicator = default_indicator if prefetch: prefetched_modules = [] for module in self.__modules(): prefetched_modules.append(module) else: prefetched_modules = None self.prefetched_modules = prefetched_modules def has_module(self, module, version): module_generator = self.prefetched_modules if module_generator is None: module_generator = self.__modules() for module_name, module_version in module_generator: names_match = module == module_name module_match = names_match and (version is None or module_version == version) if module_match: return True return False def __modules(self): raw_output = self.__module_avail_output().decode("utf-8") for line in StringIO(raw_output): line = line and line.strip() if not line or line.startswith("-"): continue line_modules = line.split() for module in line_modules: if module.endswith(self.default_indicator): module = module[0:-len(self.default_indicator)].strip() module_parts = module.split('/') module_version = None if len(module_parts) == 2: module_version = module_parts[1] module_name = module_parts[0] yield module_name, module_version def __module_avail_output(self): avail_command = [self.module_dependency_resolver.modulecmd, 'sh', 'avail'] return Popen(avail_command, stderr=PIPE, env={'MODULEPATH': self.modulepath}).communicate()[1] class ModuleDependency(Dependency): """Converts module dependencies into shell expressions using modulecmd. Using Environment Modules' 'modulecmd' (specifically 'modulecmd sh load') to convert module specifications into shell expressions for inclusion in the script used to run a tool in Galaxy.""" dict_collection_visible_keys = Dependency.dict_collection_visible_keys + ['module_name', 'module_version', 'dependency_resolver'] dependency_type = 'module' def __init__(self, module_dependency_resolver, module_name, module_version=None, exact=True, dependency_resolver=None): self.module_dependency_resolver = module_dependency_resolver self.module_name = module_name self.module_version = module_version self._exact = exact self.dependency_resolver = dependency_resolver @property def name(self): return self.module_name @property def version(self): return self.module_version @property def exact(self): return self._exact def shell_commands(self): module_to_load = self.module_name if self.module_version: module_to_load = f'{self.module_name}/{self.module_version}' command = 'MODULEPATH={}; export MODULEPATH; eval `{} sh load {}`'.format(self.module_dependency_resolver.modulepath, self.module_dependency_resolver.modulecmd, module_to_load) return command def _string_as_bool(value): return str(value).lower() == "true" __all__ = ('ModuleDependencyResolver', )