Source code for galaxy.tool_shed.util.tool_util

import logging
import os
import shutil

import galaxy.tools
from galaxy import util
from galaxy.datatypes.sniff import is_column_based
from galaxy.tool_shed.util import basic_util
from galaxy.util import checkers
from galaxy.util.expressions import ExpressionContext
from galaxy.web.form_builder import SelectField

log = logging.getLogger(__name__)


[docs]def build_shed_tool_conf_select_field(app):
    """Build a SelectField whose options are the keys in app.toolbox.shed_tool_confs."""
    options = []
    for dynamic_tool_conf_filename in app.toolbox.dynamic_conf_filenames():
        if dynamic_tool_conf_filename.startswith("./"):
            option_label = dynamic_tool_conf_filename.replace("./", "", 1)
        else:
            option_label = dynamic_tool_conf_filename
        options.append((option_label, dynamic_tool_conf_filename))
    select_field = SelectField(name="shed_tool_conf")
    for option_tup in options:
        select_field.add_option(option_tup[0], option_tup[1])
    return select_field


[docs]def build_tool_panel_section_select_field(app):
    """Build a SelectField whose options are the sections of the current in-memory toolbox."""
    options = []
    for section_id, section_name in app.toolbox.get_sections():
        options.append((section_name, section_id))
    select_field = SelectField(name="tool_panel_section_id", field_id="tool_panel_section_select")
    for option_tup in options:
        select_field.add_option(option_tup[0], option_tup[1])
    return select_field


[docs]def copy_sample_file(app, filename, dest_path=None):
    """
    Copies a sample file at `filename` to `the dest_path`
    directory and strips the '.sample' extensions from `filename`.
    Returns the path to the copied file (with the .sample extension).
    """
    if dest_path is None:
        dest_path = os.path.abspath(app.config.tool_data_path)
    sample_file_name = basic_util.strip_path(filename)
    copied_file = sample_file_name.rsplit(".sample", 1)[0]
    full_source_path = os.path.abspath(filename)
    full_destination_path = os.path.join(dest_path, sample_file_name)
    # Don't copy a file to itself - not sure how this happens, but sometimes it does...
    if full_source_path != full_destination_path:
        # It's ok to overwrite the .sample version of the file.
        shutil.copy(full_source_path, full_destination_path)
    # Only create the .loc file if it does not yet exist.  We don't overwrite it in case it
    # contains stuff proprietary to the local instance.
    non_sample_path = os.path.join(dest_path, copied_file)
    if not os.path.lexists(non_sample_path):
        shutil.copy(full_source_path, os.path.join(dest_path, copied_file))
    return non_sample_path


[docs]def copy_sample_files(app, sample_files, tool_path=None, sample_files_copied=None, dest_path=None):
    """
    Copy all appropriate files to dest_path in the local Galaxy environment that have not
    already been copied.  Those that have been copied are contained in sample_files_copied.
    The default value for dest_path is ~/tool-data.  We need to be careful to copy only
    appropriate files here because tool shed repositories can contain files ending in .sample
    that should not be copied to the ~/tool-data directory.
    """
    filenames_not_to_copy = ["tool_data_table_conf.xml.sample"]
    sample_files_copied = util.listify(sample_files_copied)
    for filename in sample_files:
        filename_sans_path = os.path.split(filename)[1]
        if filename_sans_path not in filenames_not_to_copy and filename not in sample_files_copied:
            if tool_path:
                filename = os.path.join(tool_path, filename)
            # Attempt to ensure we're copying an appropriate file.
            if is_data_index_sample_file(filename):
                copy_sample_file(app, filename, dest_path=dest_path)


[docs]def generate_message_for_invalid_tools(
    app, invalid_file_tups, repository, metadata_dict, as_html=True, displaying_invalid_tool=False
):
    if as_html:
        new_line = "<br/>"
        bold_start = "<b>"
        bold_end = "</b>"
    else:
        new_line = "\n"
        bold_start = ""
        bold_end = ""
    message = ""
    if app.name == "galaxy":
        tip_rev = str(repository.changeset_revision)
    else:
        tip_rev = str(repository.tip())
    if not displaying_invalid_tool:
        if metadata_dict:
            message += f"Metadata may have been defined for some items in revision '{tip_rev}'.  "
            message += f"Correct the following problems if necessary and reset metadata.{new_line}"
        else:
            message += f"Metadata cannot be defined for revision '{tip_rev}' so this revision cannot be automatically "
            message += (
                f"installed into a local Galaxy instance.  Correct the following problems and reset metadata.{new_line}"
            )
    for itc_tup in invalid_file_tups:
        tool_file, exception_msg = itc_tup
        if exception_msg.find("No such file or directory") >= 0:
            exception_items = exception_msg.split()
            missing_file_items = exception_items[7].split("/")
            missing_file = missing_file_items[-1].rstrip("'")
            if missing_file.endswith(".loc"):
                sample_ext = f"{missing_file}.sample"
            else:
                sample_ext = missing_file
            correction_msg = f"This file refers to a missing file {bold_start}{missing_file}{bold_end}.  "
            correction_msg += (
                f"Upload a file named {bold_start}{sample_ext}{bold_end} to the repository to correct this error."
            )
        else:
            if as_html:
                correction_msg = exception_msg
            else:
                correction_msg = (
                    exception_msg.replace("<br/>", new_line).replace("<b>", bold_start).replace("</b>", bold_end)
                )
        message += f"{bold_start}{tool_file}{bold_end} - {correction_msg}{new_line}"
    return message


[docs]def get_tool_path_install_dir(partial_install_dir, shed_tool_conf_dict, tool_dict, config_elems):
    for elem in config_elems:
        if elem.tag == "tool":
            if elem.get("guid") == tool_dict["guid"]:
                tool_path = shed_tool_conf_dict["tool_path"]
                relative_install_dir = os.path.join(tool_path, partial_install_dir)
                return tool_path, relative_install_dir
        elif elem.tag == "section":
            for section_elem in elem:
                if section_elem.tag == "tool":
                    if section_elem.get("guid") == tool_dict["guid"]:
                        tool_path = shed_tool_conf_dict["tool_path"]
                        relative_install_dir = os.path.join(tool_path, partial_install_dir)
                        return tool_path, relative_install_dir
    return None, None


[docs]def handle_missing_index_file(app, tool_path, sample_files, repository_tools_tups, sample_files_copied):
    """
    Inspect each tool to see if it has any input parameters that are dynamically
    generated select lists that depend on a .loc file.  This method is not called
    from the tool shed, but from Galaxy when a repository is being installed.
    """
    for repository_tools_tup in repository_tools_tups:
        tup_path, guid, repository_tool = repository_tools_tup
        params_with_missing_index_file = repository_tool.params_with_missing_index_file
        for param in params_with_missing_index_file:
            options = param.options
            missing_file_name = basic_util.strip_path(options.missing_index_file)
            if missing_file_name not in sample_files_copied:
                # The repository must contain the required xxx.loc.sample file.
                for sample_file in sample_files:
                    sample_file_name = basic_util.strip_path(sample_file)
                    if sample_file_name == f"{missing_file_name}.sample":
                        target_path = copy_sample_file(app, os.path.join(tool_path, sample_file))
                        if options.tool_data_table and options.tool_data_table.missing_index_file:
                            options.tool_data_table.handle_found_index_file(target_path)
                        sample_files_copied.append(target_path)
                        break
    return repository_tools_tups, sample_files_copied


[docs]def is_data_index_sample_file(file_path):
    """
    Attempt to determine if a .sample file is appropriate for copying to ~/tool-data when
    a tool shed repository is being installed into a Galaxy instance.
    """
    # Currently most data index files are tabular, so check that first.  We'll assume that
    # if the file is tabular, it's ok to copy.
    if is_column_based(file_path):
        return True
    # If the file is any of the following, don't copy it.
    if checkers.check_html(file_path):
        return False
    if checkers.check_image(file_path):
        return False
    if checkers.check_binary(name=file_path):
        return False
    if checkers.is_bz2(file_path):
        return False
    if checkers.is_gzip(file_path):
        return False
    if checkers.is_zip(file_path):
        return False
    # Default to copying the file if none of the above are true.
    return True


[docs]def new_state(trans, tool, invalid=False):
    """Create a new `DefaultToolState` for the received tool.  Only inputs on the first page will be initialized."""
    state = galaxy.tools.DefaultToolState()
    state.inputs = {}
    if invalid:
        # We're attempting to display a tool in the tool shed that has been determined to have errors, so is invalid.
        return state
    try:
        # Attempt to generate the tool state using the standard Galaxy-side code
        return tool.new_state(trans)
    except Exception as e:
        # Fall back to building tool state as below
        log.debug(
            'Failed to build tool state for tool "%s" using standard method, will try to fall back on custom method: %s',
            tool.id,
            e,
        )
    inputs = tool.inputs_by_page[0]
    context = ExpressionContext(state.inputs, parent=None)
    for input in inputs.values():
        try:
            state.inputs[input.name] = input.get_initial_value(trans, context)
        except Exception:
            # FIXME: not all values should be an empty list
            state.inputs[input.name] = []
    return state


[docs]def panel_entry_per_tool(tool_section_dict):
    # Return True if tool_section_dict looks like this.
    # {<Tool guid> :
    #    [{ tool_config : <tool_config_file>,
    #       id: <ToolSection id>,
    #       version : <ToolSection version>,
    #       name : <TooSection name>}]}
    # But not like this.
    # { id: <ToolSection id>, version : <ToolSection version>, name : <TooSection name>}
    if not tool_section_dict:
        return False
    if len(tool_section_dict) != 3:
        return True
    for k in tool_section_dict.keys():
        if k not in ["id", "version", "name"]:
            return True
    return False


__all__ = (
    "build_shed_tool_conf_select_field",
    "build_tool_panel_section_select_field",
    "copy_sample_file",
    "copy_sample_files",
    "generate_message_for_invalid_tools",
    "get_tool_path_install_dir",
    "handle_missing_index_file",
    "is_data_index_sample_file",
    "new_state",
    "panel_entry_per_tool",
)