Warning
This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.
Source code for galaxy.tool_shed.galaxy_install.datatypes.custom_datatype_manager
import logging
import os
from galaxy.tool_shed.util.basic_util import strip_path
from galaxy.tool_shed.util.hg_util import get_config_from_disk
from galaxy.tool_shed.util.shed_util_common import DATATYPES_CONFIG_FILENAME
from galaxy.util.tool_shed.xml_util import parse_xml
log = logging.getLogger(__name__)
[docs]class CustomDatatypeLoader:
[docs] def alter_config_and_load_prorietary_datatypes(self, datatypes_config, relative_install_dir,
deactivate=False, override=True):
"""
Parse a custom datatypes config (a datatypes_conf.xml file included in an installed
tool shed repository) and add information to appropriate element attributes that will
enable custom datatype class modules, datatypes converters and display applications
to be discovered and properly imported by the datatypes registry. The value of override
will be False when a tool shed repository is being installed. Since installation is
occurring after the datatypes registry has been initialized, the registry's contents
cannot be overridden by conflicting data types.
"""
tree, error_message = parse_xml(datatypes_config)
if tree is None:
return None, None
datatypes_config_root = tree.getroot()
registration = datatypes_config_root.find('registration')
if registration is None:
# We have valid XML, but not a valid custom datatypes definition.
return None, None
converter_path, display_path = self.get_converter_and_display_paths(registration,
relative_install_dir)
if converter_path:
# Path to datatype converters
registration.attrib['proprietary_converter_path'] = converter_path
if display_path:
# Path to datatype display applications
registration.attrib['proprietary_display_path'] = display_path
relative_path_to_datatype_file_name = None
datatype_files = datatypes_config_root.find('datatype_files')
datatype_class_modules = []
if datatype_files is not None:
# The <datatype_files> tag set contains any number of <datatype_file> tags.
# <datatype_files>
# <datatype_file name="gmap.py"/>
# <datatype_file name="metagenomics.py"/>
# </datatype_files>
# We'll add attributes to the datatype tag sets so that the modules can be properly imported
# by the datatypes registry.
for elem in datatype_files.findall('datatype_file'):
datatype_file_name = elem.get('name', None)
if datatype_file_name:
# Find the file in the installed repository.
for root, _dirs, files in os.walk(relative_install_dir):
if root.find('.hg') < 0:
for name in files:
if name == datatype_file_name:
datatype_class_modules.append(os.path.join(root, name))
break
break
if datatype_class_modules:
for relative_path_to_datatype_file_name in datatype_class_modules:
datatype_file_name_path, datatype_file_name = os.path.split(relative_path_to_datatype_file_name)
for elem in registration.findall('datatype'):
# Handle 'type' attribute which should be something like one of the following:
# type="gmap:GmapDB"
# type="galaxy.datatypes.gmap:GmapDB"
dtype = elem.get('type', None)
if dtype:
fields = dtype.split(':')
proprietary_datatype_module = fields[0]
if proprietary_datatype_module.find('.') >= 0:
# Handle the case where datatype_module is "galaxy.datatypes.gmap".
proprietary_datatype_module = proprietary_datatype_module.split('.')[-1]
# The value of proprietary_path must be an absolute path due to job_working_directory.
elem.attrib['proprietary_path'] = os.path.abspath(datatype_file_name_path)
elem.attrib['proprietary_datatype_module'] = proprietary_datatype_module
# Load custom datatypes
self.app.datatypes_registry.load_datatypes(root_dir=self.app.config.root,
config=datatypes_config_root,
deactivate=deactivate,
override=override)
return converter_path, display_path
[docs] def create_repository_dict_for_proprietary_datatypes(self, tool_shed, name, owner, installed_changeset_revision,
tool_dicts, converter_path=None, display_path=None):
return dict(tool_shed=tool_shed,
repository_name=name,
repository_owner=owner,
installed_changeset_revision=installed_changeset_revision,
tool_dicts=tool_dicts,
converter_path=converter_path,
display_path=display_path)
[docs] def get_converter_and_display_paths(self, registration_elem, relative_install_dir):
"""
Find the relative path to data type converters and display applications included
in installed tool shed repositories.
"""
converter_path = None
display_path = None
for elem in registration_elem.findall('datatype'):
if not converter_path:
# If any of the <datatype> tag sets contain <converter> tags, set the converter_path
# if it is not already set. This requires developers to place all converters in the
# same subdirectory within the repository hierarchy.
for converter in elem.findall('converter'):
converter_config = converter.get('file', None)
if converter_config:
converter_config_file_name = strip_path(converter_config)
for root, _dirs, files in os.walk(relative_install_dir):
if root.find('.hg') < 0:
for name in files:
if name == converter_config_file_name:
# The value of converter_path must be absolute due to job_working_directory.
converter_path = os.path.abspath(root)
break
if converter_path:
break
if not display_path:
# If any of the <datatype> tag sets contain <display> tags, set the display_path
# if it is not already set. This requires developers to place all display acpplications
# in the same subdirectory within the repository hierarchy.
for display_app in elem.findall('display'):
display_config = display_app.get('file', None)
if display_config:
display_config_file_name = strip_path(display_config)
for root, _dirs, files in os.walk(relative_install_dir):
if root.find('.hg') < 0:
for name in files:
if name == display_config_file_name:
# The value of display_path must be absolute due to job_working_directory.
display_path = os.path.abspath(root)
break
if display_path:
break
if converter_path and display_path:
break
return converter_path, display_path
[docs] def load_installed_datatype_converters(self, installed_repository_dict, deactivate=False):
"""Load or deactivate proprietary datatype converters."""
self.app.datatypes_registry.load_datatype_converters(self.app.toolbox,
installed_repository_dict=installed_repository_dict,
deactivate=deactivate)
[docs] def load_installed_datatypes(self, repository, relative_install_dir, deactivate=False):
"""
Load proprietary datatypes and return information needed for loading custom
datatypes converters and display applications later.
"""
metadata = repository.metadata
repository_dict = None
datatypes_config = get_config_from_disk(DATATYPES_CONFIG_FILENAME, relative_install_dir)
if datatypes_config:
converter_path, display_path = \
self.alter_config_and_load_prorietary_datatypes(datatypes_config,
relative_install_dir,
deactivate=deactivate)
if converter_path or display_path:
# Create a dictionary of tool shed repository related information.
repository_dict = \
self.create_repository_dict_for_proprietary_datatypes(tool_shed=repository.tool_shed,
name=repository.name,
owner=repository.owner,
installed_changeset_revision=repository.installed_changeset_revision,
tool_dicts=metadata.get('tools', []),
converter_path=converter_path,
display_path=display_path)
return repository_dict
[docs] def load_installed_display_applications(self, installed_repository_dict, deactivate=False):
"""Load or deactivate custom datatype display applications."""
self.app.datatypes_registry.load_display_applications(self.app, installed_repository_dict=installed_repository_dict,
deactivate=deactivate)