-
Loraine Gueguen authoredadd phaeoexplorer_constants.py. Add constants. Add SpeciesData attributes. Update run_tool_and_download_single_output_dataset(). Move install_repository_revision() to utilities.py. Refactor run_wf (WIP)
d44121bf
run_workflow_phaeoexplorer.py 100.81 KiB
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import bioblend
import bioblend.galaxy.objects
import argparse
import os
import logging
import sys
import json
import time
from bioblend.galaxy.objects import GalaxyInstance
from bioblend import galaxy
import utilities
import speciesData
import constants
import phaoexplorer_constants
"""
gga_init.py
Usage: $ python3 gga_init.py -i input_example.yml --config [config file] [OPTIONS]
"""
class RunWorkflow(speciesData.SpeciesData):
"""
Run a workflow into the galaxy instance's history of a given species
This script is made to work for a Phaeoexplorer-specific workflow, but can be adapted to run any workflow,
provided the user creates their own workflow in a .ga format, and change the set_parameters function
to have the correct parameters for their workflow
"""
def __init__(self, parameters_dictionary):
super().__init__(parameters_dictionary)
self.chado_species_name = " ".join(utilities.filter_empty_not_empty_items(
[self.species, self.strain, self.sex])["not_empty"])
self.abbreviation = self.genus_uppercase[0] + ". " + self.chado_species_name
self.common = self.name
if not self.common_name is None and self.common_name != "":
self.common = self.common_name
self.history_name = str(self.genus_species)
self.genome_analysis_name = "genome v{0} of {1}".format(self.genome_version, self.full_name)
self.genome_analysis_programversion = "genome v{0}".format(self.genome_version)
self.genome_analysis_sourcename = self.full_name
self.ogs_analysis_name = "OGS{0} of {1}".format(self.ogs_version, self.full_name)
self.ogs_analysis_programversion = "OGS{0}".format(self.ogs_version)
self.ogs_analysis_sourcename = self.full_name
def set_history(self):
"""
Create or set the working history to the current species one
:return:
"""
try:
histories = self.instance.histories.get_histories(name=self.history_name)
self.history_id = histories[0]["id"]
logging.debug("History ID set for {0}: {1}".format(self.history_name, self.history_id))
except IndexError:
logging.info("Creating history for %s" % self.history_name)
history = self.instance.histories.create_history(name=self.history_name)
self.history_id = history["id"]
logging.debug("History ID set for {0}: {1}".format(self.history_name, self.history_id))
return self.history_id
def set_galaxy_instance(self):
"""
Test the connection to the galaxy instance for the current organism
Exit if we cannot connect to the instance
"""
logging.debug("Connecting to the galaxy instance (%s)" % self.instance_url)
self.instance = galaxy.GalaxyInstance(url=self.instance_url,
email=self.config[constants.CONF_GALAXY_DEFAULT_ADMIN_EMAIL],
password=self.config[constants.CONF_GALAXY_DEFAULT_ADMIN_PASSWORD]
)
try:
self.instance.histories.get_histories()
except bioblend.ConnectionError:
logging.critical("Cannot connect to galaxy instance (%s) " % self.instance_url)
sys.exit()
else:
logging.debug("Successfully connected to galaxy instance (%s) " % self.instance_url)
return self.instance
def install_changesets_revisions_for_individual_tools(self):
"""
This function is used to verify that installed tools called outside workflows have the correct versions and changesets
If it finds versions don't match, will install the correct version + changeset in the instance
Doesn't do anything if versions match
:return:
"""
self.set_galaxy_instance()
logging.info("Validating installed individual tools versions and changesets")
# Verify that the add_organism and add_analysis versions are correct in the instance
add_organism_tool = self.instance.tools.show_tool(phaoexplorer_constants.ADD_ORGANISM_TOOL_ID)
add_analysis_tool = self.instance.tools.show_tool(phaoexplorer_constants.ADD_ANALYSIS_TOOL_ID)
get_organisms_tool = self.instance.tools.show_tool(phaoexplorer_constants.GET_ORGANISMS_TOOL_ID)
get_analyses_tool = self.instance.tools.show_tool(phaoexplorer_constants.GET_ANALYSES_TOOL_ID)
analysis_sync_tool = self.instance.tools.show_tool(phaoexplorer_constants.ANALYSIS_SYNC_TOOL_ID)
organism_sync_tool = self.instance.tools.show_tool(phaoexplorer_constants.ORGANISM_SYNC_TOOL_ID)
# changeset for 2.3.4+galaxy0 has to be manually found because there is no way to get the wanted changeset of a non installed tool via bioblend
# except for workflows (.ga) that already contain the changeset revisions inside the steps ids
utilities.install_repository_revision(current_version=get_organisms_tool["version"],
toolshed_dict=get_organisms_tool["tool_shed_repository"],
version_to_install=phaoexplorer_constants.GET_ORGANISMS_TOOL_VERSION,
changeset_revision=phaoexplorer_constants.GET_ORGANISMS_TOOL_CHANGESET_REVISION,
instance=self.instance)
utilities.install_repository_revision(current_version=get_analyses_tool["version"],
toolshed_dict=get_analyses_tool["tool_shed_repository"],
version_to_install=phaoexplorer_constants.GET_ANALYSES_TOOL_VERSION,
changeset_revision=phaoexplorer_constants.GET_ANALYSES_TOOL_CHANGESET_REVISION,
instance=self.instance)
utilities.install_repository_revision(current_version=add_organism_tool["version"],
toolshed_dict=add_organism_tool["tool_shed_repository"],
version_to_install=phaoexplorer_constants.ADD_ORGANISM_TOOL_VERSION,
changeset_revision=phaoexplorer_constants.ADD_ORGANISM_TOOL_CHANGESET_REVISION,
instance=self.instance)
utilities.install_repository_revision(current_version=add_analysis_tool["version"],
toolshed_dict=add_analysis_tool["tool_shed_repository"],
version_to_install=phaoexplorer_constants.ADD_ANALYSIS_TOOL_VERSION,
changeset_revision=phaoexplorer_constants.ADD_ANALYSIS_TOOL_CHANGESET_REVISION,
instance=self.instance)
utilities.install_repository_revision(current_version=analysis_sync_tool["version"],
toolshed_dict=analysis_sync_tool["tool_shed_repository"],
version_to_install=phaoexplorer_constants.ANALYSIS_SYNC_TOOL_VERSION,
changeset_revision=phaoexplorer_constants.ANALYSIS_SYNC_TOOL_CHANGESET_REVISION,
instance=self.instance)
utilities.install_repository_revision(current_version=organism_sync_tool["version"],
toolshed_dict=organism_sync_tool["tool_shed_repository"],
version_to_install=phaoexplorer_constants.ORGANISM_SYNC_TOOL_VERSION,
changeset_revision=phaoexplorer_constants.ORGANISM_SYNC_TOOL_CHANGESET_REVISION,
instance=self.instance)
logging.info("Success: individual tools versions and changesets validated")
def add_analysis(self, name, programversion, sourcename):
add_analysis_tool_dataset = utilities.run_tool_and_download_single_output_dataset(
instance=self.instance,
tool_id=phaoexplorer_constants.ADD_ANALYSIS_TOOL_ID,
history_id=self.history_id,
tool_inputs={"name": name,
"program": phaoexplorer_constants.ADD_ANALYSIS_TOOL_PARAM_PROGRAM,
"programversion": programversion,
"sourcename": sourcename,
"date_executed": phaoexplorer_constants.ADD_ANALYSIS_TOOL_PARAM_DATE})
analysis_dict = json.loads(add_analysis_tool_dataset)
analysis_id = str(analysis_dict["analysis_id"])
return analysis_id
def sync_analysis(self, analysis_id):
time.sleep(60)
utilities.run_tool(
instance=self.instance,
tool_id=phaoexplorer_constants.ANALYSIS_SYNC_TOOL_ID,
history_id=self.history_id,
tool_inputs={"analysis_id": analysis_id})
def add_organism_and_sync(self):
get_organisms_tool_dataset = utilities.run_tool_and_download_single_output_dataset(
instance=self.instance,
tool_id=phaoexplorer_constants.GET_ORGANISMS_TOOL_ID,
history_id=self.history_id,
tool_inputs={},
time_sleep=10
)
organisms_dict_list = json.loads(get_organisms_tool_dataset) # Turn the dataset into a list for parsing
org_id = None
# Look up list of outputs (dictionaries)
for org_dict in organisms_dict_list:
if org_dict["genus"] == self.genus_uppercase and org_dict["species"] == self.chado_species_name:
org_id = str(org_dict["organism_id"]) # id needs to be a str to be recognized by chado tools
if org_id is None:
add_organism_tool_dataset = utilities.run_tool_and_download_single_output_dataset(
instance=self.instance,
tool_id=phaoexplorer_constants.ADD_ORGANISM_TOOL_ID,
history_id=self.history_id,
tool_inputs={"abbr": self.abbreviation,
"genus": self.genus_uppercase,
"species": self.chado_species_name,
"common": self.common})
organism_dict = json.loads(add_organism_tool_dataset)
org_id = str(organism_dict["organism_id"]) # id needs to be a str to be recognized by chado tools
# Synchronize newly added organism in Tripal
logging.info("Synchronizing organism %s in Tripal" % self.full_name)
time.sleep(60)
utilities.run_tool(
instance=self.instance,
tool_id=phaoexplorer_constants.ORGANISM_SYNC_TOOL_ID,
history_id=self.history_id,
tool_inputs={"organism_id": org_id})
return org_id
def get_analyses(self):
get_analyses_tool_dataset = utilities.run_tool_and_download_single_output_dataset(
instance=self.instance,
tool_id=phaoexplorer_constants.GET_ANALYSES_TOOL_ID,
history_id=self.history_id,
tool_inputs={},
time_sleep=10
)
analyses_dict_list = json.loads(get_analyses_tool_dataset)
return analyses_dict_list
def add_analysis_and_sync(self, analyses_dict_list, analysis_name, analysis_programversion, analysis_sourcename):
"""
Add one analysis to Chado database
Required for Chado Load Tripal Synchronize workflow (which should be ran as the first workflow)
Called outside workflow for practical reasons (Chado add doesn't have an input link for analysis or organism)
"""
analysis_id = None
# Look up list of outputs (dictionaries)
for analyses_dict in analyses_dict_list:
if analyses_dict["name"] == analysis_name:
analysis_id = str(analyses_dict["analysis_id"])
if analysis_id is None:
analysis_id = self.add_analysis(
name=analysis_name,
programversion=analysis_programversion,
sourcename=analysis_sourcename
)
# Synchronize analysis in Tripal
logging.info("Synchronizing analysis %s in Tripal" % analysis_name)
self.sync_analysis(analysis_id=analysis_id)
return(analysis_id)
def add_organism_blastp_analysis(self):
"""
Add OGS and genome vX analyses to Chado database
Required for Chado Load Tripal Synchronize workflow (which should be ran as the first workflow)
Called outside workflow for practical reasons (Chado add doesn't have an input link for analysis or organism)
:return:
"""
self.set_galaxy_instance()
self.set_history()
tool_version = "2.3.4+galaxy0"
get_organism_tool = self.instance.tools.show_tool("toolshed.g2.bx.psu.edu/repos/gga/chado_organism_get_organisms/organism_get_organisms/2.3.4+galaxy0")
get_organisms = self.instance.tools.run_tool(
tool_id="toolshed.g2.bx.psu.edu/repos/gga/chado_organism_get_organisms/organism_get_organisms/%s" % tool_version,
history_id=self.history_id,
tool_inputs={})
time.sleep(10) # Ensure the tool has had time to complete
org_outputs = get_organisms["outputs"] # Outputs from the get_organism tool
org_job_out_id = org_outputs[0]["id"] # ID of the get_organism output dataset (list of dicts)
org_json_output = self.instance.datasets.download_dataset(dataset_id=org_job_out_id) # Download the dataset
org_output = json.loads(org_json_output) # Turn the dataset into a list for parsing
org_id = None
# Look up list of outputs (dictionaries)
for organism_output_dict in org_output:
if organism_output_dict["genus"] == self.genus and organism_output_dict["species"] == "{0} {1}".format(self.species, self.sex):
correct_organism_id = str(organism_output_dict["organism_id"]) # id needs to be a str to be recognized by chado tools
org_id = str(correct_organism_id)
if org_id is None:
add_org_job = self.instance.tools.run_tool(
tool_id="toolshed.g2.bx.psu.edu/repos/gga/chado_organism_add_organism/organism_add_organism/%s" % tool_version,
history_id=self.history_id,
tool_inputs={"abbr": self.abbreviation,
"genus": self.genus_uppercase,
"species": self.chado_species_name,
"common": self.common})
org_job_out_id = add_org_job["outputs"][0]["id"]
org_json_output = self.instance.datasets.download_dataset(dataset_id=org_job_out_id)
org_output = json.loads(org_json_output)
org_id = str(org_output["organism_id"]) # id needs to be a str to be recognized by chado tools
# Synchronize newly added organism in Tripal
logging.info("Synchronizing organism %s in Tripal" % self.full_name)
time.sleep(60)
org_sync = self.instance.tools.run_tool(tool_id="toolshed.g2.bx.psu.edu/repos/gga/tripal_organism_sync/organism_sync/3.2.1.0",
history_id=self.history_id,
tool_inputs={"organism_id": org_id})
get_analyses = self.instance.tools.run_tool(
tool_id="toolshed.g2.bx.psu.edu/repos/gga/chado_analysis_get_analyses/analysis_get_analyses/%s" % tool_version,
history_id=self.history_id,
tool_inputs={})
time.sleep(10)
analysis_outputs = get_analyses["outputs"]
analysis_job_out_id = analysis_outputs[0]["id"]
analysis_json_output = self.instance.datasets.download_dataset(dataset_id=analysis_job_out_id)
analysis_output = json.loads(analysis_json_output)
blastp_analysis_id = None
# Look up list of outputs (dictionaries)
for analysis_output_dict in analysis_output:
if analysis_output_dict["name"] == "Diamond on " + self.full_name_lowercase + " OGS" + self.ogs_version:
blastp_analysis_id = str(analysis_output_dict["analysis_id"])
if blastp_analysis_id is None:
add_blast_analysis_job = self.instance.tools.run_tool(
tool_id="toolshed.g2.bx.psu.edu/repos/gga/chado_analysis_add_analysis/analysis_add_analysis/%s" % tool_version,
history_id=self.history_id,
tool_inputs={"name": "Diamond on " + self.full_name_lowercase + " OGS" + self.ogs_version,
"program": "Performed by Genoscope",
"programversion": str(self.sex + " OGS" + self.ogs_version),
"sourcename": "Genoscope",
"date_executed": self.date})
analysis_outputs = add_blast_analysis_job["outputs"]
analysis_job_out_id = analysis_outputs[0]["id"]
analysis_json_output = self.instance.datasets.download_dataset(dataset_id=analysis_job_out_id)
analysis_output = json.loads(analysis_json_output)
blastp_analysis_id = str(analysis_output["analysis_id"])
# Synchronize blastp analysis
logging.info("Synchronizing Diamong blastp OGS%s analysis in Tripal" % self.ogs_version)
time.sleep(60)
blastp_analysis_sync = self.instance.tools.run_tool(tool_id="toolshed.g2.bx.psu.edu/repos/gga/tripal_analysis_sync/analysis_sync/3.2.1.0",
history_id=self.history_id,
tool_inputs={"analysis_id": blastp_analysis_id})
# print({"org_id": org_id, "genome_analysis_id": genome_analysis_id, "ogs_analysis_id": ogs_analysis_id})
return({"org_id": org_id, "blastp_analysis_id": blastp_analysis_id})
def add_organism_interproscan_analysis(self):
"""
Add OGS and genome vX analyses to Chado database
Required for Chado Load Tripal Synchronize workflow (which should be ran as the first workflow)
Called outside workflow for practical reasons (Chado add doesn't have an input link for analysis or organism)
:return:
"""
self.set_galaxy_instance()
self.set_history()
tool_version = "2.3.4+galaxy0"
get_organism_tool = self.instance.tools.show_tool("toolshed.g2.bx.psu.edu/repos/gga/chado_organism_get_organisms/organism_get_organisms/2.3.4+galaxy0")
get_organisms = self.instance.tools.run_tool(
tool_id="toolshed.g2.bx.psu.edu/repos/gga/chado_organism_get_organisms/organism_get_organisms/%s" % tool_version,
history_id=self.history_id,
tool_inputs={})
time.sleep(10) # Ensure the tool has had time to complete
org_outputs = get_organisms["outputs"] # Outputs from the get_organism tool
org_job_out_id = org_outputs[0]["id"] # ID of the get_organism output dataset (list of dicts)
org_json_output = self.instance.datasets.download_dataset(dataset_id=org_job_out_id) # Download the dataset
org_output = json.loads(org_json_output) # Turn the dataset into a list for parsing
org_id = None
# Look up list of outputs (dictionaries)
for organism_output_dict in org_output:
if organism_output_dict["genus"] == self.genus and organism_output_dict["species"] == "{0} {1}".format(self.species, self.sex):
correct_organism_id = str(organism_output_dict["organism_id"]) # id needs to be a str to be recognized by chado tools
org_id = str(correct_organism_id)
if org_id is None:
add_org_job = self.instance.tools.run_tool(
tool_id="toolshed.g2.bx.psu.edu/repos/gga/chado_organism_add_organism/organism_add_organism/%s" % tool_version,
history_id=self.history_id,
tool_inputs={"abbr": self.abbreviation,
"genus": self.genus_uppercase,
"species": self.chado_species_name,
"common": self.common})
org_job_out_id = add_org_job["outputs"][0]["id"]
org_json_output = self.instance.datasets.download_dataset(dataset_id=org_job_out_id)
org_output = json.loads(org_json_output)
org_id = str(org_output["organism_id"]) # id needs to be a str to be recognized by chado tools
# Synchronize newly added organism in Tripal
logging.info("Synchronizing organism %s in Tripal" % self.full_name)
time.sleep(60)
org_sync = self.instance.tools.run_tool(tool_id="toolshed.g2.bx.psu.edu/repos/gga/tripal_organism_sync/organism_sync/3.2.1.0",
history_id=self.history_id,
tool_inputs={"organism_id": org_id})
get_analyses = self.instance.tools.run_tool(
tool_id="toolshed.g2.bx.psu.edu/repos/gga/chado_analysis_get_analyses/analysis_get_analyses/%s" % tool_version,
history_id=self.history_id,
tool_inputs={})
time.sleep(10)
analysis_outputs = get_analyses["outputs"]
analysis_job_out_id = analysis_outputs[0]["id"]
analysis_json_output = self.instance.datasets.download_dataset(dataset_id=analysis_job_out_id)
analysis_output = json.loads(analysis_json_output)
interpro_analysis_id = None
# Look up list of outputs (dictionaries)
for analysis_output_dict in analysis_output:
if analysis_output_dict["name"] == "Interproscan on " + self.full_name_lowercase + " OGS" + self.ogs_version:
interpro_analysis_id = str(analysis_output_dict["analysis_id"])
if interpro_analysis_id is None:
add_interproscan_analysis_job = self.instance.tools.run_tool(
tool_id="toolshed.g2.bx.psu.edu/repos/gga/chado_analysis_add_analysis/analysis_add_analysis/%s" % tool_version,
history_id=self.history_id,
tool_inputs={"name": "Interproscan on " + self.full_name_lowercase + " OGS" + self.ogs_version,
"program": "Performed by Genoscope",
"programversion": str(self.sex + " OGS" + self.ogs_version),
"sourcename": "Genoscope",
"date_executed": self.date})
analysis_outputs = add_interproscan_analysis_job["outputs"]
analysis_job_out_id = analysis_outputs[0]["id"]
analysis_json_output = self.instance.datasets.download_dataset(dataset_id=analysis_job_out_id)
analysis_output = json.loads(analysis_json_output)
interpro_analysis_id = str(analysis_output["analysis_id"])
# Synchronize blastp analysis
logging.info("Synchronizing Diamong blastp OGS%s analysis in Tripal" % self.ogs_version)
time.sleep(60)
interproscan_analysis_sync = self.instance.tools.run_tool(tool_id="toolshed.g2.bx.psu.edu/repos/gga/tripal_analysis_sync/analysis_sync/3.2.1.0",
history_id=self.history_id,
tool_inputs={"analysis_id": interpro_analysis_id})
# print({"org_id": org_id, "genome_analysis_id": genome_analysis_id, "ogs_analysis_id": ogs_analysis_id})
return({"org_id": org_id, "interpro_analysis_id": interpro_analysis_id})
def get_interpro_analysis_id(self):
"""
"""
# Get interpro ID
interpro_analysis = self.instance.tools.run_tool(
tool_id="toolshed.g2.bx.psu.edu/repos/gga/chado_analysis_get_analyses/analysis_get_analyses/2.3.4+galaxy0",
history_id=self.history_id,
tool_inputs={"name": "InterproScan on OGS%s" % self.ogs_version})
interpro_analysis_job_out = interpro_analysis["outputs"][0]["id"]
interpro_analysis_json_output = self.instance.datasets.download_dataset(dataset_id=interpro_analysis_job_out)
try:
interpro_analysis_output = json.loads(interpro_analysis_json_output)[0]
self.interpro_analysis_id = str(interpro_analysis_output["analysis_id"])
except IndexError as exc:
logging.critical("No matching InterproScan analysis exists in the instance's chado database")
sys.exit(exc)
return self.interpro_analysis_id
def get_invocation_report(self, workflow_name):
"""
Debugging method for workflows
Simply logs and returns a report of the previous workflow invocation (execution of a workflow in
the instance via the API)
:param workflow_name:
:return:
"""
workflow_attributes = self.instance.workflows.get_workflows(name=workflow_name)
workflow_id = workflow_attributes[1]["id"] # Most recently imported workflow (index 1 in the list)
invocations = self.instance.workflows.get_invocations(workflow_id=workflow_id)
invocation_id = invocations[1]["id"] # Most recent invocation
invocation_report = self.instance.invocations.get_invocation_report(invocation_id=invocation_id)
logging.debug(invocation_report)
return invocation_report
def import_datasets_into_history(self):
"""
Find datasets in a library, get their ID and import them into the current history if they are not already
:return:
"""
# Instanciate the instance
gio = GalaxyInstance(url=self.instance_url,
email=self.config[constants.CONF_GALAXY_DEFAULT_ADMIN_EMAIL],
password=self.config[constants.CONF_GALAXY_DEFAULT_ADMIN_PASSWORD])
prj_lib = gio.libraries.get_previews(name=constants.GALAXY_LIBRARY_NAME)
library_id = prj_lib[0].id
folder_dict_list = self.instance.libraries.get_folders(library_id=str(library_id))
folders_id_dict = {}
# Loop over the folders in the library and map folders names to their IDs
for folder_dict in folder_dict_list:
folders_id_dict[folder_dict["name"]] = folder_dict["id"]
# Iterating over the folders to find datasets and map datasets to their IDs
for folder_name, folder_id in folders_id_dict.items():
if folder_name == "/genome/{0}/v{1}".format(self.species_folder_name, self.genome_version):
sub_folder_content = self.instance.folders.show_folder(folder_id=folder_id, contents=True)
for k2, v2 in sub_folder_content.items():
for e in v2:
if type(e) == dict:
if e["name"].endswith(".fasta"):
self.datasets["genome_file"] = e["ldda_id"]
self.datasets_name["genome_file"] = e["name"]
if folder_name == "/annotation/{0}/OGS{1}".format(self.species_folder_name, self.ogs_version):
sub_folder_content = self.instance.folders.show_folder(folder_id=folder_id, contents=True)
for k2, v2 in sub_folder_content.items():
for e in v2:
if type(e) == dict:
if "transcripts" in e["name"]:
self.datasets["transcripts_file"] = e["ldda_id"]
self.datasets_name["transcripts_file"] = e["name"]
elif "proteins" in e["name"]:
self.datasets["proteins_file"] = e["ldda_id"]
self.datasets_name["proteins_file"] = e["name"]
elif "gff" in e["name"]:
self.datasets["gff_file"] = e["ldda_id"]
self.datasets_name["gff_file"] = e["name"]
elif "interpro" in e["name"]:
self.datasets["interproscan_file"] = e["ldda_id"]
self.datasets_name["interproscan_file"] = e["name"]
elif "blastp" in e["name"]:
self.datasets["blastp_file"] = e["ldda_id"]
self.datasets_name["blastp_file"] = e["name"]
history_datasets_li = self.instance.datasets.get_datasets()
genome_hda_id, gff_hda_id, transcripts_hda_id, proteins_hda_id, blastp_hda_id, interproscan_hda_id = None, None, None, None, None, None
# Finding datasets in history (matching datasets names)
for dataset in history_datasets_li:
dataset_name = dataset["name"]
dataset_id = dataset["id"]
if dataset_name == "{0}_v{1}.fasta".format(self.dataset_prefix, self.genome_version):
genome_hda_id = dataset_id
if dataset_name == "{0}_OGS{1}_{2}.gff".format(self.dataset_prefix, self.ogs_version, self.date):
gff_hda_id = dataset_id
if dataset_name == "{0}_OGS{1}_transcripts.fasta".format(self.dataset_prefix, self.ogs_version):
transcripts_hda_id = dataset_id
if dataset_name == "{0}_OGS{1}_proteins.fasta".format(self.dataset_prefix, self.ogs_version):
proteins_hda_id = dataset_id
if dataset_name == "{0}_OGS{1}_blastp.xml".format(self.dataset_prefix, self.ogs_version):
blastp_hda_id = dataset_id
if dataset_name == "{0}_OGS{1}_interproscan.xml".format(self.dataset_prefix, self.ogs_version):
interproscan_hda_id = dataset_id
# Import each dataset into history if it is not imported
logging.debug("Uploading datasets into history %s" % self.history_id)
if genome_hda_id is None:
genome_dataset_upload = self.instance.histories.upload_dataset_from_library(history_id=self.history_id, lib_dataset_id=self.datasets["genome_file"])
genome_hda_id = genome_dataset_upload["id"]
if gff_hda_id is None:
gff_dataset_upload = self.instance.histories.upload_dataset_from_library(history_id=self.history_id, lib_dataset_id=self.datasets["gff_file"])
gff_hda_id = gff_dataset_upload["id"]
if transcripts_hda_id is None:
transcripts_dataset_upload = self.instance.histories.upload_dataset_from_library(history_id=self.history_id, lib_dataset_id=self.datasets["transcripts_file"])
transcripts_hda_id = transcripts_dataset_upload["id"]
if proteins_hda_id is None:
proteins_dataset_upload = self.instance.histories.upload_dataset_from_library(history_id=self.history_id, lib_dataset_id=self.datasets["proteins_file"])
proteins_hda_id = proteins_dataset_upload["id"]
if interproscan_hda_id is None:
try:
interproscan_dataset_upload = self.instance.histories.upload_dataset_from_library(history_id=self.history_id, lib_dataset_id=self.datasets["interproscan_file"])
interproscan_hda_id = interproscan_dataset_upload["id"]
except Exception as exc:
logging.debug("Interproscan file not found in library (history: {0})".format(self.history_id))
if blastp_hda_id is None:
try:
blastp_dataset_upload = self.instance.histories.upload_dataset_from_library(history_id=self.history_id, lib_dataset_id=self.datasets["blastp_file"])
blastp_hda_id = blastp_dataset_upload["id"]
except Exception as exc:
logging.debug("blastp file not found in library (history: {0})".format(self.history_id))
# logging.debug("History dataset IDs (hda_id) for %s:" % self.full_name)
# logging.debug({"genome_hda_id": genome_hda_id,
# "gff_hda_id": gff_hda_id,
# "transcripts_hda_id": transcripts_hda_id,
# "proteins_hda_id": proteins_hda_id,
# "blastp_hda_id": blastp_hda_id,
# "interproscan_hda_id": interproscan_hda_id})
# Return a dict made of the hda ids
return {"genome_hda_id": genome_hda_id,
"gff_hda_id": gff_hda_id,
"transcripts_hda_id": transcripts_hda_id,
"proteins_hda_id": proteins_hda_id,
"blastp_hda_id": blastp_hda_id,
"interproscan_hda_id": interproscan_hda_id}
def run_workflow(workflow_path, workflow_parameters, datamap, config, input_species_number):
"""
Run a workflow in galaxy
Requires the .ga file to be loaded as a dictionary (optionally could be uploaded as a raw file)
:param workflow_name:
:param workflow_parameters:
:param datamap:
:return:
"""
logging.info("Importing workflow %s" % str(workflow_path))
# Load the workflow file (.ga) in a buffer
with open(workflow_path, 'r') as ga_in_file:
# Then store the decoded json dictionary
workflow_dict = json.load(ga_in_file)
# In case of the Jbrowse workflow, we unfortunately have to manually edit the parameters instead of setting them
# as runtime values, using runtime parameters makes the tool throw an internal critical error ("replace not found" error)
# Scratchgmod test: need "http" (or "https"), the hostname (+ port)
if "jbrowse_menu_url" not in config.keys():
jbrowse_menu_url = "https://{hostname}/sp/{genus_sp}/feature/{Genus}/{species}/mRNA/{id}".format(hostname=self.config["hostname"], genus_sp=self.genus_species, Genus=self.genus_uppercase, species=self.species, id="{id}")
else:
jbrowse_menu_url = config["jbrowse_menu_url"]
if workflow_name == "Jbrowse":
workflow_dict["steps"]["2"]["tool_state"] = workflow_dict["steps"]["2"]["tool_state"].replace("__MENU_URL__", jbrowse_menu_url)
# The UNIQUE_ID is specific to a combination genus_species_strain_sex so every combination should have its unique workflow
# in galaxy --> define a naming method for these workflows
workflow_dict["steps"]["3"]["tool_state"] = workflow_dict["steps"]["3"]["tool_state"].replace("__FULL_NAME__", self.full_name).replace("__UNIQUE_ID__", self.species_folder_name)
# Import the workflow in galaxy as a dict
self.instance.workflows.import_workflow_dict(workflow_dict=workflow_dict)
# Get its attributes
workflow_attributes = self.instance.workflows.get_workflows(name=workflow_name)
# Then get its ID (required to invoke the workflow)
workflow_id = workflow_attributes[0]["id"] # Index 0 is the most recently imported workflow (the one we want)
show_workflow = self.instance.workflows.show_workflow(workflow_id=workflow_id)
# Check if the workflow is found
try:
logging.debug("Workflow ID: %s" % workflow_id)
except bioblend.ConnectionError:
logging.warning("Error retrieving workflow attributes for workflow %s" % workflow_name)
# Finally, invoke the workflow alogn with its datamap, parameters and the history in which to invoke it
self.instance.workflows.invoke_workflow(workflow_id=workflow_id,
history_id=self.history_id,
params=workflow_parameters,
inputs=datamap,
allow_tool_state_corrections=True)
logging.info("Successfully imported and invoked workflow {0}, check the galaxy instance ({1}) for the jobs state".format(workflow_name, self.instance_url))
def create_sp_workflow_dict(sp_dict, main_dir, config, workflow_type):
"""
"""
sp_workflow_dict = {}
run_workflow_for_current_organism = RunWorkflow(parameters_dictionary=sp_dict)
# Verifying the galaxy container is running
if utilities.check_galaxy_state(network_name=run_workflow_for_current_organism.genus_species,
script_dir=run_workflow_for_current_organism.script_dir):
# Setting some of the instance attributes
run_workflow_for_current_organism.main_dir = main_dir
run_workflow_for_current_organism.species_dir = os.path.join(run_workflow_for_current_organism.main_dir,
run_workflow_for_current_organism.genus_species +
"/")
# Parse the config yaml file
run_workflow_for_current_organism.config = config
# Set the instance url attribute --> TODO: the localhost rule in the docker-compose still doesn't work on scratchgmodv1
run_workflow_for_current_organism.instance_url = "http://localhost:{0}/sp/{1}/galaxy/".format(
run_workflow_for_current_organism.config[constants.CONF_ALL_HTTP_PORT],
run_workflow_for_current_organism.genus_species)
if workflow_type == phaoexplorer_constants.WORKFLOW_LOAD_FASTA_GFF_JBROWSE:
run_workflow_for_current_organism.set_galaxy_instance()
history_id = run_workflow_for_current_organism.set_history()
run_workflow_for_current_organism.install_changesets_revisions_for_individual_tools()
analyses_dict_list = run_workflow_for_current_organism.get_analyses()
org_id = run_workflow_for_current_organism.add_organism_and_sync()
genome_analysis_id = run_workflow_for_current_organism.add_analysis_and_sync(
analyses_dict_list=analyses_dict_list,
analysis_name=run_workflow_for_current_organism.genome_analysis_name,
analysis_programversion=run_workflow_for_current_organism.genome_analysis_programversion,
analysis_sourcename=run_workflow_for_current_organism.genome_analysis_sourcename
)
ogs_analysis_id = run_workflow_for_current_organism.add_analysis_and_sync(
analyses_dict_list=analyses_dict_list,
analysis_name=run_workflow_for_current_organism.ogs_analysis_name,
analysis_programversion=run_workflow_for_current_organism.ogs_analysis_programversion,
analysis_sourcename=run_workflow_for_current_organism.ogs_analysis_sourcename
)
hda_ids = run_workflow_for_current_organism.import_datasets_into_history()
# Create the dictionary holding all attributes needed to connect to the galaxy instance
attributes = {"genus": run_workflow_for_current_organism.genus,
"species": run_workflow_for_current_organism.species,
"genus_species": run_workflow_for_current_organism.genus_species,
"full_name": run_workflow_for_current_organism.full_name,
"species_folder_name": run_workflow_for_current_organism.species_folder_name,
"sex": run_workflow_for_current_organism.sex,
"strain": run_workflow_for_current_organism.strain,
"org_id": org_id,
"genome_analysis_id": genome_analysis_id,
"ogs_analysis_id": ogs_analysis_id,
"hda_ids": hda_ids,
"history_id": history_id,
"instance": run_workflow_for_current_organism.instance,
"instance_url": run_workflow_for_current_organism.instance_url,
"email": config[constants.CONF_GALAXY_DEFAULT_ADMIN_EMAIL],
"password": config[constants.CONF_GALAXY_DEFAULT_ADMIN_PASSWORD]}
sp_workflow_dict[run_workflow_for_current_organism.genus_species] = {run_workflow_for_current_organism.genus_species.strain_sex: attributes}
else:
logging.critical("The galaxy container for %s is not ready yet!" % run_workflow_for_current_organism.genus_species)
sys.exit()
return sp_workflow_dict
if workflow_type == "blast":
run_workflow_for_current_organism.set_galaxy_instance()
history_id = run_workflow_for_current_organism.set_history()
run_workflow_for_current_organism.install_changesets_revisions_for_individual_tools()
ids = run_workflow_for_current_organism.add_organism_blastp_analysis()
org_id = ids["org_id"]
blastp_analysis_id = ids["blastp_analysis_id"]
hda_ids = run_workflow_for_current_organism.import_datasets_into_history()
strain_sex = "{0}_{1}".format(run_workflow_for_current_organism.strain, run_workflow_for_current_organism.sex)
genus_species = run_workflow_for_current_organism.genus_species
# Create the dictionary holding all attributes needed to connect to the galaxy instance
attributes = {"genus": run_workflow_for_current_organism.genus,
"species": run_workflow_for_current_organism.species,
"genus_species": run_workflow_for_current_organism.genus_species,
"full_name": run_workflow_for_current_organism.full_name,
"species_folder_name": run_workflow_for_current_organism.species_folder_name,
"sex": run_workflow_for_current_organism.sex,
"strain": run_workflow_for_current_organism.strain,
"org_id": org_id,
"blastp_analysis_id": blastp_analysis_id,
"hda_ids": hda_ids,
"history_id": history_id,
"instance": run_workflow_for_current_organism.instance,
"instance_url": run_workflow_for_current_organism.instance_url,
"email": config[constants.CONF_GALAXY_DEFAULT_ADMIN_EMAIL],
"password": config[constants.CONF_GALAXY_DEFAULT_ADMIN_PASSWORD]}
sp_workflow_dict[genus_species] = {strain_sex: attributes}
if workflow_type == "interpro":
run_workflow_for_current_organism.set_galaxy_instance()
history_id = run_workflow_for_current_organism.set_history()
run_workflow_for_current_organism.install_changesets_revisions_for_individual_tools()
ids = run_workflow_for_current_organism.add_organism_interproscan_analysis()
org_id = ids["org_id"]
interpro_analysis_id = ids["interpro_analysis_id"]
hda_ids = run_workflow_for_current_organism.import_datasets_into_history()
strain_sex = "{0}_{1}".format(run_workflow_for_current_organism.strain, run_workflow_for_current_organism.sex)
genus_species = run_workflow_for_current_organism.genus_species
# Create the dictionary holding all attributes needed to connect to the galaxy instance
attributes = {"genus": run_workflow_for_current_organism.genus,
"species": run_workflow_for_current_organism.species,
"genus_species": run_workflow_for_current_organism.genus_species,
"full_name": run_workflow_for_current_organism.full_name,
"species_folder_name": run_workflow_for_current_organism.species_folder_name,
"sex": run_workflow_for_current_organism.sex,
"strain": run_workflow_for_current_organism.strain,
"org_id": org_id,
"interpro_analysis_id": interpro_analysis_id,
"hda_ids": hda_ids,
"history_id": history_id,
"instance": run_workflow_for_current_organism.instance,
"instance_url": run_workflow_for_current_organism.instance_url,
"email": config[constants.CONF_GALAXY_DEFAULT_ADMIN_EMAIL],
"password": config[constants.CONF_GALAXY_DEFAULT_ADMIN_PASSWORD]}
sp_workflow_dict[genus_species] = {strain_sex: attributes}
else:
logging.critical("The galaxy container for %s is not ready yet!" % run_workflow_for_current_organism.full_name)
sys.exit()
def install_changesets_revisions_from_workflow(instance, workflow_path):
"""
Read a .ga file to extract the information about the different tools called.
Check if every tool is installed via a "show_tool".
If a tool is not installed (versions don't match), send a warning to the logger and install the required changeset (matching the tool version)
Doesn't do anything if versions match
:return:
"""
logging.info("Validating that installed tools versions and changesets match workflow versions")
# Load the workflow file (.ga) in a buffer
with open(workflow_path, 'r') as ga_in_file:
# Then store the decoded json dictionary
workflow_dict = json.load(ga_in_file)
# Look up every "step_id" looking for tools
for k, v in workflow_dict["steps"].items():
if v["tool_id"]:
# Get the descriptive dictionary of the installed tool (using the tool id in the workflow)
show_tool = instance.tools.show_tool(v["tool_id"])
# Check if an installed version matches the workflow tool version
# (If it's not installed, the show_tool version returned will be a default version with the suffix "XXXX+0")
if show_tool["version"] != v["tool_version"]:
# If it doesn't match, proceed to install of the correct changeset revision
toolshed = "https://" + v["tool_shed_repository"]["tool_shed"]
name = v["tool_shed_repository"]["name"]
owner = v["tool_shed_repository"]["owner"]
changeset_revision = v["tool_shed_repository"]["changeset_revision"]
logging.warning("Installed tool versions for tool {0} do not match the version required by the specified workflow, installing changeset {1}".format(name, changeset_revision))
# Install changeset
instance.toolshed.install_repository_revision(tool_shed_url=toolshed, name=name, owner=owner,
changeset_revision=changeset_revision,
install_tool_dependencies=True,
install_repository_dependencies=False,
install_resolver_dependencies=True)
else:
toolshed = "https://" + v["tool_shed_repository"]["tool_shed"]
name = v["tool_shed_repository"]["name"]
owner = v["tool_shed_repository"]["owner"]
changeset_revision = v["tool_shed_repository"]["changeset_revision"]
logging.debug("Installed tool versions for tool {0} match the version in the specified workflow (changeset {1})".format(name, changeset_revision))
logging.info("Tools versions and changesets from workflow validated")
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Run Galaxy workflows, specific to Phaeoexplorer data")
parser.add_argument("input",
type=str,
help="Input file (yml)")
parser.add_argument("-v", "--verbose",
help="Increase output verbosity",
action="store_true")
parser.add_argument("--config",
type=str,
help="Config path, default to the 'config' file inside the script repository")
parser.add_argument("--main-directory",
type=str,
help="Where the stack containers will be located, defaults to working directory")
parser.add_argument("--workflow", "-w",
type=str,
help="Worfklow to run. Available options: load_fasta_gff_jbrowse, blast, interpro")
args = parser.parse_args()
bioblend_logger = logging.getLogger("bioblend")
if args.verbose:
logging.basicConfig(level=logging.DEBUG)
bioblend_logger.setLevel(logging.DEBUG)
else:
logging.basicConfig(level=logging.INFO)
bioblend_logger.setLevel(logging.INFO)
# Parsing the config file if provided, using the default config otherwise
if args.config:
config_file = os.path.abspath(args.config)
else:
config_file = os.path.join(os.path.dirname(os.path.realpath(sys.argv[0])), constants.DEFAULT_CONFIG)
config = utilities.parse_config(config_file)
main_dir = None
if not args.main_directory:
main_dir = os.getcwd()
else:
main_dir = os.path.abspath(args.main_directory)
sp_dict_list = utilities.parse_input(args.input)
workflow_type = None
# Checking if user specified a workflow to run
if not args.workflow:
logging.critical("No workflow type specified, exiting")
sys.exit()
elif args.workflow in phaoexplorer_constants.WORKFLOW_VALID_TYPES:
workflow_type = args.workflow
logging.info("Workflow type set to '%s'" % workflow_type)
script_dir = os.path.dirname(os.path.realpath(sys.argv[0]))
all_sp_workflow_dict = {}
if workflow_type == phaoexplorer_constants.WORKFLOW_LOAD_FASTA_GFF_JBROWSE:
for sp_dict in sp_dict_list:
# Add and retrieve all analyses/organisms for the current input species and add their IDs to the input dictionary
current_sp_workflow_dict = create_sp_workflow_dict(
sp_dict,
main_dir=main_dir,
config=config,
workflow_type=phaoexplorer_constants.WORKFLOW_LOAD_FASTA_GFF_JBROWSE)
current_sp_key = list(current_sp_workflow_dict.keys())[0]
current_sp_value = list(current_sp_workflow_dict.values())[0]
current_sp_strain_sex_key = list(current_sp_value.keys())[0]
current_sp_strain_sex_value = list(current_sp_value.values())[0]
# Add the species dictionary to the complete dictionary
# This dictionary contains every organism present in the input file
# Its structure is the following:
# {genus species: {strain1_sex1: {variables_key: variables_values}, strain1_sex2: {variables_key: variables_values}}}
if not current_sp_key in all_sp_workflow_dict.keys():
all_sp_workflow_dict[current_sp_key] = current_sp_value
else:
all_sp_workflow_dict[current_sp_key][current_sp_strain_sex_key] = current_sp_strain_sex_value
for k, v in all_sp_workflow_dict.items():
if len(list(v.keys())) == 1:
logging.info("Input organism %s: 1 species detected in input dictionary" % k)
# Set workflow path (1 organism)
workflow_path = os.path.join(os.path.abspath(script_dir), "workflows_phaeoexplorer/Galaxy-Workflow-chado_load_tripal_synchronize_jbrowse_1org_v4.ga")
# Instance object required variables
instance_url, email, password = None, None, None
# Set the galaxy instance variables
for k2, v2 in v.items():
instance_url = v2["instance_url"]
email = v2["email"]
password = v2["password"]
instance = galaxy.GalaxyInstance(url=instance_url, email=email, password=password)
# Check if the versions of tools specified in the workflow are installed in galaxy
install_changesets_revisions_from_workflow(workflow_path=workflow_path, instance=instance)
organism_key_name = list(v.keys())
org_dict = v[organisms_key_name[0]]
# print("\n")
# print(org_dict)
history_id = org_dict["history_id"]
# Organism 1 attributes
org_genus = org_dict["genus"]
org_species = org_dict["species"]
org_genus_species = org_dict["genus_species"]
org_species_folder_name = org_dict["species_folder_name"]
org_full_name = org_dict["full_name"]
org_strain = org_dict["sex"]
org_sex = org_dict["strain"]
org_org_id = org_dict["org_id"]
org_genome_analysis_id = org_dict["genome_analysis_id"]
org_ogs_analysis_id = org_dict["ogs_analysis_id"]
org_genome_hda_id = org_dict["hda_ids"]["genome_hda_id"]
org_transcripts_hda_id = org_dict["hda_ids"]["transcripts_hda_id"]
org_proteins_hda_id = org_dict["hda_ids"]["proteins_hda_id"]
org_gff_hda_id = org_dict["hda_ids"]["gff_hda_id"]
# Store these values into a dict for parameters logging/validation
org_parameters_dict = {
"org_genus": org_genus,
"org_species": org_species,
"org_genus_species": org_genus_species,
"org_species_folder_name": org_species_folder_name,
"org_full_name": org_full_name,
"org_strain": org_strain,
"org_sex": org_sex,
"org_org_id": org_org_id,
"org_genome_analysis_id": org_genome_analysis_id,
"org_ogs_analysis_id": org_ogs_analysis_id,
"org_genome_hda_id": org_genome_hda_id,
"org_transcripts_hda_id": org_transcripts_hda_id,
"org_proteins_hda_id": org_proteins_hda_id,
"org_gff_hda_id": org_gff_hda_id,
}
# Look for empty parameters values, throw a critical error if a parameter value is invalid
for param_name, param_value in org_parameters_dict.items():
if param_value is None or param_value == "":
logging.critical("Empty parameter value found for organism {0} (parameter: {1}, parameter value: {2})".format(org_full_name, param_name, param_value))
sys.exit()
# Set the workflow parameters (individual tools runtime parameters in the workflow)
workflow_parameters = {}
GENOME_FASTA_FILE_ORG = "0"
GFF_FILE_ORG = "1"
PROTEINS_FASTA_FILE_ORG = "2"
LOAD_FASTA_ORG = "3"
JBROWSE_ORG = "4"
LOAD_GFF_ORG = "5"
JBROWSE_CONTAINER = "6"
SYNC_FEATURES_ORG = "7"
POPULATE_MAT_VIEWS = "8"
INDEX_TRIPAL_DATA = "9"
# Input files have no parameters (they are set via assigning the hda IDs in the datamap parameter of the bioblend method)
workflow_parameters[GENOME_FASTA_FILE_ORG] = {}
workflow_parameters[GFF_FILE_ORG] = {}
workflow_parameters[PROTEINS_FASTA_FILE_ORG] = {}
workflow_parameters[LOAD_FASTA_ORG] = {"organism": org_org_id,
"analysis_id": org_genome_analysis_id,
"do_update": "true"}
workflow_parameters[JBROWSE_ORG] = {}
workflow_parameters[LOAD_GFF_ORG] = {"organism": org_org_id, "analysis_id": org_ogs_analysis_id}
workflow_parameters[SYNC_FEATURES_ORG] = {"organism_id": org_org_id}
# POPULATE + INDEX DATA
workflow_parameters[POPULATE_MAT_VIEWS] = {}
workflow_parameters[INDEX_TRIPAL_DATA] = {}
# Set datamap (mapping of input files in the workflow)
datamap = {}
datamap[GENOME_FASTA_FILE_ORG] = {"src": "hda", "id": org_genome_hda_id}
datamap[GFF_FILE_ORG] = {"src": "hda", "id": org_gff_hda_id}
datamap[PROTEINS_FASTA_FILE_ORG] = {"src": "hda", "id": org_proteins_hda_id}
with open(workflow_path, 'r') as ga_in_file:
# Store the decoded json dictionary
workflow_dict = json.load(ga_in_file)
workflow_name = workflow_dict["name"]
# For the Jbrowse tool, we unfortunately have to manually edit the parameters instead of setting them
# as runtime values, using runtime parameters makes the tool throw an internal critical error ("replace not found" error)
# Scratchgmod test: need "http" (or "https"), the hostname (+ port)
jbrowse_menu_url_org = "https://{hostname}/sp/{genus_sp}/feature/{Genus}/{species}/mRNA/{id}".format(hostname=config["hostname"], genus_sp=org_genus_species, Genus=org_genus[0].upper() + org_genus[1:], species=org_species, id="{id}")
if "jbrowse_menu_url" not in config.keys():
jbrowse_menu_url_org = "https://{hostname}/sp/{genus_sp}/feature/{Genus}/{species}/mRNA/{id}".format(hostname=config["hostname"], genus_sp=org_genus_species, Genus=org_genus[0].upper() + org_genus[1:], species=org_species, id="{id}")
else:
jbrowse_menu_url_org = config["jbrowse_menu_url"] + "/sp/{genus_sp}/feature/{Genus}/{species}/mRNA/{id}".format(genus_sp=org_genus_species, Genus=org_genus[0].upper() + org_genus[1:], species=org_species, id="{id}")
# Replace values in the workflow dictionary
workflow_dict["steps"]["4"]["tool_state"] = workflow_dict["steps"]["4"]["tool_state"].replace("__MENU_URL_ORG__", jbrowse_menu_url_org)
workflow_dict["steps"]["6"]["tool_state"] = workflow_dict["steps"]["6"]["tool_state"].replace("__DISPLAY_NAME_ORG__", org_full_name).replace("__UNIQUE_ID_ORG__", org_species_folder_name)
# Import the workflow in galaxy as a dict
instance.workflows.import_workflow_dict(workflow_dict=workflow_dict)
# Get its attributes
workflow_attributes = instance.workflows.get_workflows(name=workflow_name)
# Then get its ID (required to invoke the workflow)
workflow_id = workflow_attributes[0]["id"] # Index 0 is the most recently imported workflow (the one we want)
show_workflow = instance.workflows.show_workflow(workflow_id=workflow_id)
# Check if the workflow is found
try:
logging.debug("Workflow ID: %s" % workflow_id)
except bioblend.ConnectionError:
logging.warning("Error finding workflow %s" % workflow_name)
# Finally, invoke the workflow alogn with its datamap, parameters and the history in which to invoke it
instance.workflows.invoke_workflow(workflow_id=workflow_id, history_id=history_id, params=workflow_parameters, inputs=datamap, allow_tool_state_corrections=True)
logging.info("Successfully imported and invoked workflow {0}, check the galaxy instance ({1}) for the jobs state".format(workflow_name, instance_url))
if len(list(v.keys())) == 2:
logging.info("Input organism %s: 2 species detected in input dictionary" % k)
# Set workflow path (2 organisms)
workflow_path = os.path.join(os.path.abspath(script_dir), "workflows_phaeoexplorer/Galaxy-Workflow-chado_load_tripal_synchronize_jbrowse_2org_v4.ga")
# Instance object required variables
instance_url, email, password = None, None, None
# Set the galaxy instance variables
for k2, v2 in v.items():
instance_url = v2["instance_url"]
email = v2["email"]
password = v2["password"]
instance = galaxy.GalaxyInstance(url=instance_url, email=email, password=password)
# Check if the versions of tools specified in the workflow are installed in galaxy
install_changesets_revisions_from_workflow(workflow_path=workflow_path, instance=instance)
# Get key names from the current organism (item 1 = organism 1, item 2 = organism 2)
organisms_key_names = list(v.keys())
org1_dict = v[organisms_key_names[0]]
org2_dict = v[organisms_key_names[1]]
history_id = org1_dict["history_id"]
# Organism 1 attributes
org1_genus = org1_dict["genus"]
org1_species = org1_dict["species"]
org1_genus_species = org1_dict["genus_species"]
org1_species_folder_name = org1_dict["species_folder_name"]
org1_full_name = org1_dict["full_name"]
org1_strain = org1_dict["sex"]
org1_sex = org1_dict["strain"]
org1_org_id = org1_dict["org_id"]
org1_genome_analysis_id = org1_dict["genome_analysis_id"]
org1_ogs_analysis_id = org1_dict["ogs_analysis_id"]
org1_genome_hda_id = org1_dict["hda_ids"]["genome_hda_id"]
org1_transcripts_hda_id = org1_dict["hda_ids"]["transcripts_hda_id"]
org1_proteins_hda_id = org1_dict["hda_ids"]["proteins_hda_id"]
org1_gff_hda_id = org1_dict["hda_ids"]["gff_hda_id"]
# Store these values into a dict for parameters logging/validation
org1_parameters_dict = {
"org1_genus": org1_genus,
"org1_species": org1_species,
"org1_genus_species": org1_genus_species,
"org1_species_folder_name": org1_species_folder_name,
"org1_full_name": org1_full_name,
"org1_strain": org1_strain,
"org1_sex": org1_sex,
"org1_org_id": org1_org_id,
"org1_genome_analysis_id": org1_genome_analysis_id,
"org1_ogs_analysis_id": org1_ogs_analysis_id,
"org1_genome_hda_id": org1_genome_hda_id,
"org1_transcripts_hda_id": org1_transcripts_hda_id,
"org1_proteins_hda_id": org1_proteins_hda_id,
"org1_gff_hda_id": org1_gff_hda_id,
}
# Look for empty parameters values, throw a critical error if a parameter value is invalid
for param_name, param_value in org1_parameters_dict.items():
if param_value is None or param_value == "":
logging.critical("Empty parameter value found for organism {0} (parameter: {1}, parameter value: {2})".format(org1_full_name, param_name, param_value))
sys.exit()
# Organism 2 attributes
org2_genus = org2_dict["genus"]
org2_species = org2_dict["species"]
org2_genus_species = org2_dict["genus_species"]
org2_species_folder_name = org2_dict["species_folder_name"]
org2_full_name = org2_dict["full_name"]
org2_strain = org2_dict["sex"]
org2_sex = org2_dict["strain"]
org2_org_id = org2_dict["org_id"]
org2_genome_analysis_id = org2_dict["genome_analysis_id"]
org2_ogs_analysis_id = org2_dict["ogs_analysis_id"]
org2_genome_hda_id = org2_dict["hda_ids"]["genome_hda_id"]
org2_transcripts_hda_id = org2_dict["hda_ids"]["transcripts_hda_id"]
org2_proteins_hda_id = org2_dict["hda_ids"]["proteins_hda_id"]
org2_gff_hda_id = org2_dict["hda_ids"]["gff_hda_id"]
# Store these values into a dict for parameters logging/validation
org2_parameters_dict = {
"org2_genus": org2_genus,
"org2_species": org2_species,
"org2_genus_species": org2_genus_species,
"org2_species_folder_name": org2_species_folder_name,
"org2_full_name": org2_full_name,
"org2_strain": org2_strain,
"org2_sex": org2_sex,
"org2_org_id": org2_org_id,
"org2_genome_analysis_id": org2_genome_analysis_id,
"org2_ogs_analysis_id": org2_ogs_analysis_id,
"org2_genome_hda_id": org2_genome_hda_id,
"org2_transcripts_hda_id": org2_transcripts_hda_id,
"org2_proteins_hda_id": org2_proteins_hda_id,
"org2_gff_hda_id": org2_gff_hda_id,
}
# Look for empty parameters values, throw a critical error if a parameter value is invalid
for param_name, param_value in org2_parameters_dict.items():
if param_value is None or param_value == "":
logging.critical("Empty parameter value found for organism {0} (parameter: {1}, parameter value: {2})".format(org2_full_name, param_name, param_value))
sys.exit()
# Source files association (ordered by their IDs in the workflow)
# WARNING: Be very careful about how the workflow is "organized" (i.e the order of the steps/datasets, check the .ga if there is any error)
GFF_FILE_ORG1 = "0"
GENOME_FASTA_FILE_ORG1 = "1"
PROTEINS_FASTA_FILE_ORG1 = "2"
GENOME_FASTA_FILE_ORG2 = "3"
GFF_FILE_ORG2 = "4"
PROTEINS_FASTA_FILE_ORG2 = "5"
LOAD_FASTA_ORG1 = "6"
JBROWSE_ORG1 = "7"
JRBOWSE_ORG2 = "8"
LOAD_GFF_ORG1 = "9"
JBROWSE_CONTAINER = "10"
SYNC_FEATURES_ORG1 = "11"
LOAD_FASTA_ORG2 = "12"
LOAD_GFF_ORG2 = "13"
SYNC_FEATURES_ORG2 = "14"
POPULATE_MAT_VIEWS = "15"
INDEX_TRIPAL_DATA = "16"
# Set the workflow parameters (individual tools runtime parameters in the workflow)
workflow_parameters = {}
# Input files have no parameters (they are set via assigning the hda IDs in the datamap parameter of the bioblend method)
workflow_parameters[GENOME_FASTA_FILE_ORG1] = {}
workflow_parameters[GFF_FILE_ORG1] = {}
workflow_parameters[PROTEINS_FASTA_FILE_ORG1] = {}
workflow_parameters[GENOME_FASTA_FILE_ORG2] = {}
workflow_parameters[GFF_FILE_ORG2] = {}
workflow_parameters[PROTEINS_FASTA_FILE_ORG2] = {}
# Organism 1
workflow_parameters[LOAD_FASTA_ORG1] = {"organism": org1_org_id,
"analysis_id": org1_genome_analysis_id,
"do_update": "true"}
# workflow_parameters[JBROWSE_ORG1] = {"jbrowse_menu_url": jbrowse_menu_url_org1}
workflow_parameters[JBROWSE_ORG1] = {}
workflow_parameters[LOAD_GFF_ORG1] = {"organism": org1_org_id, "analysis_id": org1_ogs_analysis_id}
workflow_parameters[SYNC_FEATURES_ORG1] = {"organism_id": org1_org_id}
# workflow_parameters[JBROWSE_CONTAINER] = {"organisms": [{"name": org1_full_name, "unique_id": org1_species_folder_name, }, {"name": org2_full_name, "unique_id": org2_species_folder_name}]}
workflow_parameters[JBROWSE_CONTAINER] = {}
# Organism 2
workflow_parameters[LOAD_FASTA_ORG2] = {"organism": org2_org_id,
"analysis_id": org2_genome_analysis_id,
"do_update": "true"}
workflow_parameters[LOAD_GFF_ORG2] = {"organism": org2_org_id, "analysis_id": org2_ogs_analysis_id}
# workflow_parameters[JRBOWSE_ORG2] = {"jbrowse_menu_url": jbrowse_menu_url_org2}
workflow_parameters[JRBOWSE_ORG2] = {}
workflow_parameters[SYNC_FEATURES_ORG2] = {"organism_id": org2_org_id}
# POPULATE + INDEX DATA
workflow_parameters[POPULATE_MAT_VIEWS] = {}
workflow_parameters[INDEX_TRIPAL_DATA] = {}
# Set datamap (mapping of input files in the workflow)
datamap = {}
# Organism 1
datamap[GENOME_FASTA_FILE_ORG1] = {"src": "hda", "id": org1_genome_hda_id}
datamap[GFF_FILE_ORG1] = {"src": "hda", "id": org1_gff_hda_id}
datamap[PROTEINS_FASTA_FILE_ORG1] = {"src": "hda", "id": org1_proteins_hda_id}
# Organism 2
datamap[GENOME_FASTA_FILE_ORG2] = {"src": "hda", "id": org2_genome_hda_id}
datamap[GFF_FILE_ORG2] = {"src": "hda", "id": org2_gff_hda_id}
datamap[PROTEINS_FASTA_FILE_ORG2] = {"src": "hda", "id": org2_proteins_hda_id}
with open(workflow_path, 'r') as ga_in_file:
# Store the decoded json dictionary
workflow_dict = json.load(ga_in_file)
workflow_name = workflow_dict["name"]
# For the Jbrowse tool, we unfortunately have to manually edit the parameters instead of setting them
# as runtime values, using runtime parameters makes the tool throw an internal critical error ("replace not found" error)
# Scratchgmod test: need "http" (or "https"), the hostname (+ port)
jbrowse_menu_url_org1 = "https://{hostname}/sp/{genus_sp}/feature/{Genus}/{species}/mRNA/{id}".format(hostname=config["hostname"], genus_sp=org1_genus_species, Genus=org1_genus[0].upper() + org1_genus[1:], species=org1_species, id="{id}")
jbrowse_menu_url_org2 = "https://{hostname}/sp/{genus_sp}/feature/{Genus}/{species}/mRNA/{id}".format(hostname=config["hostname"], genus_sp=org2_genus_species, Genus=org2_genus[0].upper() + org2_genus[1:], species=org2_species, id="{id}")
if "jbrowse_menu_url" not in config.keys():
jbrowse_menu_url_org1 = "https://{hostname}/sp/{genus_sp}/feature/{Genus}/{species}/mRNA/{id}".format(hostname=config["hostname"], genus_sp=org1_genus_species, Genus=org1_genus[0].upper() + org1_genus[1:], species=org1_species, id="{id}")
jbrowse_menu_url_org2 = "https://{hostname}/sp/{genus_sp}/feature/{Genus}/{species}/mRNA/{id}".format(hostname=config["hostname"], genus_sp=org2_genus_species, Genus=org2_genus[0].upper() + org2_genus[1:], species=org2_species, id="{id}")
else:
jbrowse_menu_url_org1 = config["jbrowse_menu_url"] + "/sp/{genus_sp}/feature/{Genus}/{species}/mRNA/{id}".format(genus_sp=org1_genus_species, Genus=org1_genus[0].upper() + org1_genus[1:], species=org1_species, id="{id}")
jbrowse_menu_url_org2 = config["jbrowse_menu_url"] + "/sp/{genus_sp}/feature/{Genus}/{species}/mRNA/{id}".format(genus_sp=org2_genus_species, Genus=org2_genus[0].upper() + org2_genus[1:], species=org2_species, id="{id}")
# show_tool_add_organism = instance.tools.show_tool(tool_id="toolshed.g2.bx.psu.edu/repos/gga/chado_organism_add_organism/organism_add_organism/2.3.4+galaxy0", io_details=True)
# print(show_tool_add_organism)
# show_jbrowse_tool = instance.tools.show_tool(tool_id="toolshed.g2.bx.psu.edu/repos/iuc/jbrowse/jbrowse/1.16.11+galaxy0", io_details=True)
# print(show_jbrowse_tool)
# show_jbrowse_container_tool = instance.tools.show_tool(tool_id="toolshed.g2.bx.psu.edu/repos/gga/jbrowse_to_container/jbrowse_to_container/0.5.1", io_details=True)
# print(show_jbrowse_container_tool)
# Replace values in the workflow dictionary
workflow_dict["steps"]["7"]["tool_state"] = workflow_dict["steps"]["7"]["tool_state"].replace("__MENU_URL_ORG1__", jbrowse_menu_url_org1)
workflow_dict["steps"]["8"]["tool_state"] = workflow_dict["steps"]["8"]["tool_state"].replace("__MENU_URL_ORG2__", jbrowse_menu_url_org2)
# The UNIQUE_ID is specific to a combination genus_species_strain_sex so every combination should have its unique workflow
# in galaxy --> define a naming method for these workflows
workflow_dict["steps"]["10"]["tool_state"] = workflow_dict["steps"]["10"]["tool_state"].replace("__DISPLAY_NAME_ORG1__", org1_full_name).replace("__UNIQUE_ID_ORG1__", org1_species_folder_name)
workflow_dict["steps"]["10"]["tool_state"] = workflow_dict["steps"]["10"]["tool_state"].replace("__DISPLAY_NAME_ORG2__", org2_full_name).replace("__UNIQUE_ID_ORG2__", org2_species_folder_name)
# Import the workflow in galaxy as a dict
instance.workflows.import_workflow_dict(workflow_dict=workflow_dict)
# Get its attributes
workflow_attributes = instance.workflows.get_workflows(name=workflow_name)
# Then get its ID (required to invoke the workflow)
workflow_id = workflow_attributes[0]["id"] # Index 0 is the most recently imported workflow (the one we want)
show_workflow = instance.workflows.show_workflow(workflow_id=workflow_id)
# Check if the workflow is found
try:
logging.debug("Workflow ID: %s" % workflow_id)
except bioblend.ConnectionError:
logging.warning("Error finding workflow %s" % workflow_name)
# Finally, invoke the workflow alogn with its datamap, parameters and the history in which to invoke it
instance.workflows.invoke_workflow(workflow_id=workflow_id, history_id=history_id, params=workflow_parameters, inputs=datamap, allow_tool_state_corrections=True)
logging.info("Successfully imported and invoked workflow {0}, check the galaxy instance ({1}) for the jobs state".format(workflow_name, instance_url))
if workflow_type == "blast":
for sp_dict in sp_dict_list:
# Add and retrieve all analyses/organisms for the current input species and add their IDs to the input dictionary
current_sp_workflow_dict = create_sp_workflow_dict(sp_dict, main_dir=args.main_directory, config=config, workfow_type="blast")
current_sp_key = list(current_sp_workflow_dict.keys())[0]
current_sp_value = list(current_sp_workflow_dict.values())[0]
current_sp_strain_sex_key = list(current_sp_value.keys())[0]
current_sp_strain_sex_value = list(current_sp_value.values())[0]
# Add the species dictionary to the complete dictionary
# This dictionary contains every organism present in the input file
# Its structure is the following:
# {genus species: {strain1_sex1: {variables_key: variables_values}, strain1_sex2: {variables_key: variables_values}}}
if not current_sp_key in all_sp_workflow_dict.keys():
all_sp_workflow_dict[current_sp_key] = current_sp_value
else:
all_sp_workflow_dict[current_sp_key][current_sp_strain_sex_key] = current_sp_strain_sex_value
if len(list(v.keys())) == 1:
logging.info("Input organism %s: 1 species detected in input dictionary" % k)
# Set workflow path (1 organism)
workflow_path = os.path.join(os.path.abspath(script_dir), "workflows_phaeoexplorer/Galaxy-Workflow-load_blast_results_1org_v1.ga")
# Instance object required variables
instance_url, email, password = None, None, None
# Set the galaxy instance variables
for k2, v2 in v.items():
instance_url = v2["instance_url"]
email = v2["email"]
password = v2["password"]
instance = galaxy.GalaxyInstance(url=instance_url, email=email, password=password)
# Check if the versions of tools specified in the workflow are installed in galaxy
install_changesets_revisions_from_workflow(workflow_path=workflow_path, instance=instance)
organism_key_name = list(v.keys())
org_dict = v[organisms_key_name[0]]
history_id = org_dict["history_id"]
# Organism attributes
org_genus = org_dict["genus"]
org_species = org_dict["species"]
org_genus_species = org_dict["genus_species"]
org_species_folder_name = org_dict["species_folder_name"]
org_full_name = org_dict["full_name"]
org_strain = org_dict["sex"]
org_sex = org_dict["strain"]
org_org_id = org_dict["org_id"]
org_blastp_analysis_id = org_dict["blastp_analysis_id"]
org_blastp_hda_id = org_dict["hda_ids"]["blastp_hda_id"]
# Store these values into a dict for parameters logging/validation
org_parameters_dict = {
"org_genus": org_genus,
"org_species": org_species,
"org_genus_species": org_genus_species,
"org_species_folder_name": org_species_folder_name,
"org_full_name": org_full_name,
"org_strain": org_strain,
"org_sex": org_sex,
"org_org_id": org_org_id,
"org_blast_analysis_id": org_blastp_analysis_id,
"org_blastp_hda_id": org_blastp_hda_id,
}
# Look for empty parameters values, throw a critical error if a parameter value is invalid
for param_name, param_value in org_parameters_dict.items():
if param_value is None or param_value == "":
logging.critical("Empty parameter value found for organism {0} (parameter: {1}, parameter value: {2})".format(org_full_name, param_name, param_value))
sys.exit()
BLASTP_FILE = "0"
LOAD_BLASTP_FILE = "1"
POPULATE_MAT_VIEWS = "2"
INDEX_TRIPAL_DATA = "3"
# Set the workflow parameters (individual tools runtime parameters in the workflow)
workflow_parameters = {}
workflow_parameters[BLASTP_FILE] = {}
workflow_parameters[LOAD_BLASTP_FILE] = {"analysis_id": org_blastp_analysis_id, "organism_id": org_org_id}
workflow_parameters[POPULATE_MAT_VIEWS] = {}
workflow_parameters[INDEX_TRIPAL_DATA] = {}
datamap = {}
datamap[BLASTP_FILE] = {"src": "hda", "id": org_blastp_hda_id}
with open(workflow_path, 'r') as ga_in_file:
# Store the decoded json dictionary
workflow_dict = json.load(ga_in_file)
workflow_name = workflow_dict["name"]
# Import the workflow in galaxy as a dict
instance.workflows.import_workflow_dict(workflow_dict=workflow_dict)
# Get its attributes
workflow_attributes = instance.workflows.get_workflows(name=workflow_name)
# Then get its ID (required to invoke the workflow)
workflow_id = workflow_attributes[0]["id"] # Index 0 is the most recently imported workflow (the one we want)
show_workflow = instance.workflows.show_workflow(workflow_id=workflow_id)
# Check if the workflow is found
try:
logging.debug("Workflow ID: %s" % workflow_id)
except bioblend.ConnectionError:
logging.warning("Error finding workflow %s" % workflow_name)
# Finally, invoke the workflow alogn with its datamap, parameters and the history in which to invoke it
instance.workflows.invoke_workflow(workflow_id=workflow_id, history_id=history_id, params=workflow_parameters, inputs=datamap, allow_tool_state_corrections=True)
logging.info("Successfully imported and invoked workflow {0}, check the galaxy instance ({1}) for the jobs state".format(workflow_name, instance_url))
if len(list(v.keys())) == 2:
logging.info("Input organism %s: 2 species detected in input dictionary" % k)
# Set workflow path (2 organisms)
workflow_path = os.path.join(os.path.abspath(script_dir), "workflows_phaeoexplorer/Galaxy-Workflow-load_blast_results_2org_v1.ga")
# Instance object required variables
instance_url, email, password = None, None, None
# Set the galaxy instance variables
for k2, v2 in v.items():
instance_url = v2["instance_url"]
email = v2["email"]
password = v2["password"]
instance = galaxy.GalaxyInstance(url=instance_url, email=email, password=password)
# Check if the versions of tools specified in the workflow are installed in galaxy
install_changesets_revisions_from_workflow(workflow_path=workflow_path, instance=instance)
organisms_key_names = list(v.keys())
org1_dict = v[organisms_key_names[0]]
org2_dict = v[organisms_key_names[1]]
history_id = org1_dict["history_id"]
# Organism 1 attributes
org1_genus = org1_dict["genus"]
org1_species = org1_dict["species"]
org1_genus_species = org1_dict["genus_species"]
org1_species_folder_name = org1_dict["species_folder_name"]
org1_full_name = org1_dict["full_name"]
org1_strain = org1_dict["sex"]
org1_sex = org1_dict["strain"]
org1_org_id = org1_dict["org_id"]
org1_blastp_analysis_id = org1_dict["blastp_analysis_id"]
org1_blastp_hda_id = org1_dict["hda_ids"]["blastp_hda_id"]
# Store these values into a dict for parameters logging/validation
org1_parameters_dict = {
"org1_genus": org1_genus,
"org1_species": org1_species,
"org1_genus_species": org1_genus_species,
"org1_species_folder_name": org1_species_folder_name,
"org1_full_name": org1_full_name,
"org1_strain": org1_strain,
"org1_sex": org1_sex,
"org1_org_id": org1_org_id,
"org1_blast_analysis_id": org1_blastp_analysis_id,
"org1_blastp_hda_id": org1_blastp_hda_id,
}
# Look for empty parameters values, throw a critical error if a parameter value is invalid
for param_name, param_value in org1_parameters_dict.items():
if param_value is None or param_value == "":
logging.critical("Empty parameter value found for organism {0} (parameter: {1}, parameter value: {2})".format(org1_full_name, param_name, param_value))
sys.exit()
# Organism 2 attributes
org2_genus = org2_dict["genus"]
org2_species = org2_dict["species"]
org2_genus_species = org2_dict["genus_species"]
org2_species_folder_name = org2_dict["species_folder_name"]
org2_full_name = org2_dict["full_name"]
org2_strain = org2_dict["sex"]
org2_sex = org2_dict["strain"]
org2_org_id = org2_dict["org_id"]
org2_blastp_analysis_id = org2_dict["blastp_analysis_id"]
org2_blastp_hda_id = org2_dict["hda_ids"]["blastp_hda_id"]
# Store these values into a dict for parameters logging/validation
org2_parameters_dict = {
"org2_genus": org2_genus,
"org2_species": org2_species,
"org2_genus_species": org2_genus_species,
"org2_species_folder_name": orgé_species_folder_name,
"org2_full_name": org2_full_name,
"org2_strain": org2_strain,
"org2_sex": org2_sex,
"org2_org_id": org2_org_id,
"org2_blast_analysis_id": org2_blastp_analysis_id,
"org2_blastp_hda_id": org2_blastp_hda_id,
}
# Look for empty parameters values, throw a critical error if a parameter value is invalid
for param_name, param_value in org2_parameters_dict.items():
if param_value is None or param_value == "":
logging.critical("Empty parameter value found for organism {0} (parameter: {1}, parameter value: {2})".format(org2_full_name, param_name, param_value))
sys.exit()
# Source files association (ordered by their IDs in the workflow)
# WARNING: Be very careful about how the workflow is "organized" (i.e the order of the steps/datasets, check the .ga if there is any error)
BLASTP_FILE_ORG1 = "0"
BLASTP_FILE_ORG2 = "1"
LOAD_BLASTP_FILE_ORG1 = "2"
LOAD_BLASTP_FILE_ORG1 = "3"
POPULATE_MAT_VIEWS = "4"
INDEX_TRIPAL_DATA = "5"
# Set the workflow parameters (individual tools runtime parameters in the workflow)
workflow_parameters = {}
# Input files have no parameters (they are set via assigning the hda IDs in the datamap parameter of the bioblend method)
workflow_parameters[BLASTP_FILE_ORG1] = {}
workflow_parameters[BLASTP_FILE_ORG2] = {}
# Organism 1
workflow_parameters[LOAD_BLASTP_FILE_ORG1] = {"organism_id": org1_org_id,
"analysis_id": org1_blastp_analysis_id}
# Organism 2
workflow_parameters[LOAD_BLASTP_FILE_ORG2] = {"organism_id": org2_org_id,
"analysis_id": org2_blastp_analysis_id}
workflow_parameters[POPULATE_MAT_VIEWS] = {}
workflow_parameters[INDEX_TRIPAL_DATA] = {}
# Set datamap (mapping of input files in the workflow)
datamap = {}
# Organism 1
datamap[BLASTP_FILE_ORG1] = {"src": "hda", "id": org1_blastp_hda_id}
# Organism 2
datamap[BLASTP_FILE_ORG2] = {"src": "hda", "id": org2_blastp_hda_id}
with open(workflow_path, 'r') as ga_in_file:
# Store the decoded json dictionary
workflow_dict = json.load(ga_in_file)
workflow_name = workflow_dict["name"]
# Import the workflow in galaxy as a dict
instance.workflows.import_workflow_dict(workflow_dict=workflow_dict)
# Get its attributes
workflow_attributes = instance.workflows.get_workflows(name=workflow_name)
# Then get its ID (required to invoke the workflow)
workflow_id = workflow_attributes[0]["id"] # Index 0 is the most recently imported workflow (the one we want)
show_workflow = instance.workflows.show_workflow(workflow_id=workflow_id)
# Check if the workflow is found
try:
logging.debug("Workflow ID: %s" % workflow_id)
except bioblend.ConnectionError:
logging.warning("Error finding workflow %s" % workflow_name)
# Finally, invoke the workflow alogn with its datamap, parameters and the history in which to invoke it
instance.workflows.invoke_workflow(workflow_id=workflow_id, history_id=history_id, params=workflow_parameters, inputs=datamap, allow_tool_state_corrections=True)
logging.info("Successfully imported and invoked workflow {0}, check the galaxy instance ({1}) for the jobs state".format(workflow_name, instance_url))
if workflow_type == "interpro":
for sp_dict in sp_dict_list:
# Add and retrieve all analyses/organisms for the current input species and add their IDs to the input dictionary
current_sp_workflow_dict = create_sp_workflow_dict(sp_dict, main_dir=args.main_directory, config=config, workfow_type="blast")
current_sp_key = list(current_sp_workflow_dict.keys())[0]
current_sp_value = list(current_sp_workflow_dict.values())[0]
current_sp_strain_sex_key = list(current_sp_value.keys())[0]
current_sp_strain_sex_value = list(current_sp_value.values())[0]
# Add the species dictionary to the complete dictionary
# This dictionary contains every organism present in the input file
# Its structure is the following:
# {genus species: {strain1_sex1: {variables_key: variables_values}, strain1_sex2: {variables_key: variables_values}}}
if not current_sp_key in all_sp_workflow_dict.keys():
all_sp_workflow_dict[current_sp_key] = current_sp_value
else:
all_sp_workflow_dict[current_sp_key][current_sp_strain_sex_key] = current_sp_strain_sex_value
if len(list(v.keys())) == 1:
logging.info("Input organism %s: 1 species detected in input dictionary" % k)
# Set workflow path (1 organism)
workflow_path = os.path.join(os.path.abspath(script_dir), "workflows_phaeoexplorer/Galaxy-Workflow-load_blast_results_1org_v1.ga")
# Instance object required variables
instance_url, email, password = None, None, None
# Set the galaxy instance variables
for k2, v2 in v.items():
instance_url = v2["instance_url"]
email = v2["email"]
password = v2["password"]
instance = galaxy.GalaxyInstance(url=instance_url, email=email, password=password)
# Check if the versions of tools specified in the workflow are installed in galaxy
install_changesets_revisions_from_workflow(workflow_path=workflow_path, instance=instance)
organism_key_name = list(v.keys())
org_dict = v[organisms_key_name[0]]
history_id = org_dict["history_id"]
# Organism attributes
org_genus = org_dict["genus"]
org_species = org_dict["species"]
org_genus_species = org_dict["genus_species"]
org_species_folder_name = org_dict["species_folder_name"]
org_full_name = org_dict["full_name"]
org_strain = org_dict["sex"]
org_sex = org_dict["strain"]
org_org_id = org_dict["org_id"]
org_inteproscan_analysis_id = org_dict["inteproscan_analysis_id"]
org_interproscan_hda_id = org_dict["hda_ids"]["interproscan_hda_id"]
# Store these values into a dict for parameters logging/validation
org_parameters_dict = {
"org_genus": org_genus,
"org_species": org_species,
"org_genus_species": org_genus_species,
"org_species_folder_name": org_species_folder_name,
"org_full_name": org_full_name,
"org_strain": org_strain,
"org_sex": org_sex,
"org_org_id": org_org_id,
"org_inteproscan_analysis_id": org_inteproscan_analysis_id,
"org_interproscan_hda_id": org_interproscan_hda_id,
}
# Look for empty parameters values, throw a critical error if a parameter value is invalid
for param_name, param_value in org_parameters_dict.items():
if param_value is None or param_value == "":
logging.critical("Empty parameter value found for organism {0} (parameter: {1}, parameter value: {2})".format(org_full_name, param_name, param_value))
sys.exit()
INTEPRO_FILE = "0"
LOAD_INTERPRO_FILE = "1"
POPULATE_MAT_VIEWS = "2"
INDEX_TRIPAL_DATA = "3"
# Set the workflow parameters (individual tools runtime parameters in the workflow)
workflow_parameters = {}
workflow_parameters[INTEPRO_FILE] = {}
workflow_parameters[LOAD_INTERPRO_FILE] = {"analysis_id": org_inteproscan_analysis_id, "organism_id": org_org_id}
workflow_parameters[POPULATE_MAT_VIEWS] = {}
workflow_parameters[INDEX_TRIPAL_DATA] = {}
datamap = {}
datamap[INTEPRO_FILE] = {"src": "hda", "id": org_interproscan_hda_id}
with open(workflow_path, 'r') as ga_in_file:
# Store the decoded json dictionary
workflow_dict = json.load(ga_in_file)
workflow_name = workflow_dict["name"]
# Import the workflow in galaxy as a dict
instance.workflows.import_workflow_dict(workflow_dict=workflow_dict)
# Get its attributes
workflow_attributes = instance.workflows.get_workflows(name=workflow_name)
# Then get its ID (required to invoke the workflow)
workflow_id = workflow_attributes[0]["id"] # Index 0 is the most recently imported workflow (the one we want)
show_workflow = instance.workflows.show_workflow(workflow_id=workflow_id)
# Check if the workflow is found
try:
logging.debug("Workflow ID: %s" % workflow_id)
except bioblend.ConnectionError:
logging.warning("Error finding workflow %s" % workflow_name)
# Finally, invoke the workflow alogn with its datamap, parameters and the history in which to invoke it
instance.workflows.invoke_workflow(workflow_id=workflow_id, history_id=history_id, params=workflow_parameters, inputs=datamap, allow_tool_state_corrections=True)
logging.info("Successfully imported and invoked workflow {0}, check the galaxy instance ({1}) for the jobs state".format(workflow_name, instance_url))
if len(list(v.keys())) == 2:
logging.info("Input organism %s: 2 species detected in input dictionary" % k)
# Set workflow path (2 organisms)
workflow_path = os.path.join(os.path.abspath(script_dir), "workflows_phaeoexplorer/Galaxy-Workflow-load_blast_results_2org_v1.ga")
# Instance object required variables
instance_url, email, password = None, None, None
# Set the galaxy instance variables
for k2, v2 in v.items():
instance_url = v2["instance_url"]
email = v2["email"]
password = v2["password"]
instance = galaxy.GalaxyInstance(url=instance_url, email=email, password=password)
# Check if the versions of tools specified in the workflow are installed in galaxy
install_changesets_revisions_from_workflow(workflow_path=workflow_path, instance=instance)
organisms_key_names = list(v.keys())
org1_dict = v[organisms_key_names[0]]
org2_dict = v[organisms_key_names[1]]
history_id = org1_dict["history_id"]
# Organism 1 attributes
org1_genus = org1_dict["genus"]
org1_species = org1_dict["species"]
org1_genus_species = org1_dict["genus_species"]
org1_species_folder_name = org1_dict["species_folder_name"]
org1_full_name = org1_dict["full_name"]
org1_strain = org1_dict["sex"]
org1_sex = org1_dict["strain"]
org1_org_id = org1_dict["org_id"]
org1_interproscan_analysis_id = org1_dict["interproscan_analysis_id"]
org1_interproscan_hda_id = org1_dict["hda_ids"]["interproscan_hda_id"]
# Store these values into a dict for parameters logging/validation
org1_parameters_dict = {
"org1_genus": org1_genus,
"org1_species": org1_species,
"org1_genus_species": org1_genus_species,
"org1_species_folder_name": org1_species_folder_name,
"org1_full_name": org1_full_name,
"org1_strain": org1_strain,
"org1_sex": org1_sex,
"org1_org_id": org1_org_id,
"org1_interproscan_analysis_id": org1_interproscan_analysis_id,
"org1_interproscan_hda_id": org1_interproscan_hda_id,
}
# Look for empty parameters values, throw a critical error if a parameter value is invalid
for param_name, param_value in org1_parameters_dict.items():
if param_value is None or param_value == "":
logging.critical("Empty parameter value found for organism {0} (parameter: {1}, parameter value: {2})".format(org1_full_name, param_name, param_value))
sys.exit()
# Organism 2 attributes
org2_genus = org2_dict["genus"]
org2_species = org2_dict["species"]
org2_genus_species = org2_dict["genus_species"]
org2_species_folder_name = org2_dict["species_folder_name"]
org2_full_name = org2_dict["full_name"]
org2_strain = org2_dict["sex"]
org2_sex = org2_dict["strain"]
org2_org_id = org2_dict["org_id"]
org2_interproscan_analysis_id = org2_dict["interproscan_analysis_id"]
org2_interproscan_hda_id = org2_dict["hda_ids"]["interproscan_hda_id"]
# Store these values into a dict for parameters logging/validation
org2_parameters_dict = {
"org2_genus": org2_genus,
"org2_species": org2_species,
"org2_genus_species": org2_genus_species,
"org2_species_folder_name": orgé_species_folder_name,
"org2_full_name": org2_full_name,
"org2_strain": org2_strain,
"org2_sex": org2_sex,
"org2_org_id": org2_org_id,
"org2_interproscan_analysis_id": org2_interproscan_analysis_id,
"org2_interproscan_hda_id": org2_interproscan_hda_id,
}
# Look for empty parameters values, throw a critical error if a parameter value is invalid
for param_name, param_value in org2_parameters_dict.items():
if param_value is None or param_value == "":
logging.critical("Empty parameter value found for organism {0} (parameter: {1}, parameter value: {2})".format(org2_full_name, param_name, param_value))
sys.exit()
# Source files association (ordered by their IDs in the workflow)
# WARNING: Be very careful about how the workflow is "organized" (i.e the order of the steps/datasets, check the .ga if there is any error)
INTERPRO_FILE_ORG1 = "0"
INTERPRO_FILE_ORG2 = "1"
LOAD_INTERPRO_FILE_ORG1 = "2"
LOAD_INTERPRO_FILE_ORG2 = "3"
POPULATE_MAT_VIEWS = "4"
INDEX_TRIPAL_DATA = "5"
# Set the workflow parameters (individual tools runtime parameters in the workflow)
workflow_parameters = {}
# Input files have no parameters (they are set via assigning the hda IDs in the datamap parameter of the bioblend method)
workflow_parameters[INTERPRO_FILE_ORG1] = {}
workflow_parameters[INTERPRO_FILE_ORG2] = {}
# Organism 1
workflow_parameters[LOAD_INTERPRO_FILE_ORG1] = {"organism_id": org1_org_id,
"analysis_id": org1_interproscan_analysis_id}
# Organism 2
workflow_parameters[LOAD_INTERPRO_FILE_ORG2] = {"organism_id": org2_org_id,
"analysis_id": org2_interproscan_analysis_id}
workflow_parameters[POPULATE_MAT_VIEWS] = {}
workflow_parameters[INDEX_TRIPAL_DATA] = {}
# Set datamap (mapping of input files in the workflow)
datamap = {}
# Organism 1
datamap[BLASTP_FILE_ORG1] = {"src": "hda", "id": org1_interproscan_hda_id}
# Organism 2
datamap[BLASTP_FILE_ORG2] = {"src": "hda", "id": org2_interproscan_hda_id}
with open(workflow_path, 'r') as ga_in_file:
# Store the decoded json dictionary
workflow_dict = json.load(ga_in_file)
workflow_name = workflow_dict["name"]
# Import the workflow in galaxy as a dict
instance.workflows.import_workflow_dict(workflow_dict=workflow_dict)
# Get its attributes
workflow_attributes = instance.workflows.get_workflows(name=workflow_name)
# Then get its ID (required to invoke the workflow)
workflow_id = workflow_attributes[0]["id"] # Index 0 is the most recently imported workflow (the one we want)
show_workflow = instance.workflows.show_workflow(workflow_id=workflow_id)
# Check if the workflow is found
try:
logging.debug("Workflow ID: %s" % workflow_id)
except bioblend.ConnectionError:
logging.warning("Error finding workflow %s" % workflow_name)
# Finally, invoke the workflow alogn with its datamap, parameters and the history in which to invoke it
instance.workflows.invoke_workflow(workflow_id=workflow_id, history_id=history_id, params=workflow_parameters, inputs=datamap, allow_tool_state_corrections=True)
logging.info("Successfully imported and invoked workflow {0}, check the galaxy instance ({1}) for the jobs state".format(workflow_name, instance_url))