Newer
Older

Arthur Le Bars
committed
#!/usr/bin/env python3

Arthur Le Bars
committed
import bioblend.galaxy.objects
import argparse
import os
import logging
import sys

Arthur Le Bars
committed
import json

Arthur Le Bars
committed
import time

Arthur Le Bars
committed

Arthur Le Bars
committed
from bioblend import galaxy

Arthur Le Bars
committed
import utilities
import utilities_bioblend

Arthur Le Bars
committed
import speciesData

Loraine Gueguen
committed
import constants
import constants_phaeo

Arthur Le Bars
committed

Arthur Le Bars
committed
"""

Arthur Le Bars
committed

Arthur Le Bars
committed
Usage: $ python3 gga_init.py -i input_example.yml --config [config file] [OPTIONS]

Arthur Le Bars
committed
"""
def __init__(self, genus_species, strain_sex, genus_uppercase, chado_species_name, full_name, species_folder_name,
org_id, history_id, instance, genome_analysis_id=None, ogs_analysis_id=None, blastp_analysis_id=None, interpro_analysis_id=None,
genome_hda_id=None, gff_hda_id=None, transcripts_hda_id=None, proteins_hda_id=None, blastp_hda_id=None, blastx_hda_id=None, interproscan_hda_id=None):
self.genus_species = genus_species
self.strain_sex = strain_sex
self.genus_uppercase = genus_uppercase
self.chado_species_name = chado_species_name,
self.full_name = full_name
self.species_folder_name = species_folder_name
self.org_id = org_id
self.genome_analysis_id = genome_analysis_id
self.ogs_analysis_id = ogs_analysis_id
self.blastp_analysis_id = blastp_analysis_id
self.interpro_analysis_id = interpro_analysis_id
self.history_id = history_id
self.instance = instance
self.genome_hda_id = genome_hda_id,
self.gff_hda_id = gff_hda_id,
self.transcripts_hda_id = transcripts_hda_id,
self.proteins_hda_id = proteins_hda_id,
self.blastp_hda_id = blastp_hda_id,
self.blastx_hda_id = blastx_hda_id,
self.interproscan_hda_id = interproscan_hda_id,
def check_param_for_workflow_load_fasta_gff_jbrowse(self):
params = [self.genus_species, self.strain_sex, self.genus_uppercase,
self.chado_species_name, self.full_name,
self.history_id, self.instance,
self.genome_analysis_id, self.ogs_analysis_id,
self.genome_hda_id, self.gff_hda_id, self.transcripts_hda_id, self.proteins_hda_id]
utilities_bioblend.check_wf_param(self.full_name, params)
def check_param_for_workflow_blastp(self):
params = [self.genus_species, self.strain_sex, self.genus_uppercase,
self.chado_species_name, self.full_name,
self.history_id, self.instance,
self.blastp_analysis_id,
self.blastp_hda_id]
utilities_bioblend.check_wf_param(self.full_name, params)
def check_param_for_workflow_interpro(self):
params = [self.genus_species, self.strain_sex, self.genus_uppercase,
self.chado_species_name, self.full_name,
self.history_id, self.instance,
self.interpro_analysis_id,
self.interproscan_hda_id]
utilities_bioblend.check_wf_param(self.full_name, params)

Arthur Le Bars
committed
"""
Run a workflow into the galaxy instance's history of a given species

Arthur Le Bars
committed

Arthur Le Bars
committed
This script is made to work for a Phaeoexplorer-specific workflow, but can be adapted to run any workflow,

Arthur Le Bars
committed
provided the user creates their own workflow in a .ga format, and change the set_parameters function

Arthur Le Bars
committed
to have the correct parameters for their workflow

Arthur Le Bars
committed

Arthur Le Bars
committed
"""

Loraine Gueguen
committed
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
self.genome_hda_id = None
self.gff_hda_id = None
self.transcripts_hda_id = None
self.proteins_hda_id = None
self.blastp_hda_id = None
self.blastx_hda_id = None
self.interproscan_hda_id = None
def install_changesets_revisions_for_individual_tools(self):

Arthur Le Bars
committed
"""
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:
"""
logging.info("Validating installed individual tools versions and changesets")

Arthur Le Bars
committed

Loraine Gueguen
committed
# Verify that the add_organism and add_analysis versions are correct in the instance
# 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

Arthur Le Bars
committed
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
utilities_bioblend.install_repository_revision(tool_id=constants_phaeo.GET_ORGANISMS_TOOL_ID,
version=constants_phaeo.GET_ORGANISMS_TOOL_VERSION,
changeset_revision=constants_phaeo.GET_ORGANISMS_TOOL_CHANGESET_REVISION,
instance=self.instance)
utilities_bioblend.install_repository_revision(tool_id=constants_phaeo.GET_ANALYSES_TOOL_ID,
version=constants_phaeo.GET_ANALYSES_TOOL_VERSION,
changeset_revision=constants_phaeo.GET_ANALYSES_TOOL_CHANGESET_REVISION,
instance=self.instance)
utilities_bioblend.install_repository_revision(tool_id=constants_phaeo.ADD_ORGANISM_TOOL_ID,
version=constants_phaeo.ADD_ORGANISM_TOOL_VERSION,
changeset_revision=constants_phaeo.ADD_ORGANISM_TOOL_CHANGESET_REVISION,
instance=self.instance)
utilities_bioblend.install_repository_revision(tool_id=constants_phaeo.ADD_ANALYSIS_TOOL_ID,
version=constants_phaeo.ADD_ANALYSIS_TOOL_VERSION,
changeset_revision=constants_phaeo.ADD_ANALYSIS_TOOL_CHANGESET_REVISION,
instance=self.instance)
utilities_bioblend.install_repository_revision(tool_id=constants_phaeo.ANALYSIS_SYNC_TOOL_ID,
version=constants_phaeo.ANALYSIS_SYNC_TOOL_VERSION,
changeset_revision=constants_phaeo.ANALYSIS_SYNC_TOOL_CHANGESET_REVISION,
instance=self.instance)
utilities_bioblend.install_repository_revision(tool_id=constants_phaeo.ORGANISM_SYNC_TOOL_ID,
version=constants_phaeo.ORGANISM_SYNC_TOOL_VERSION,
changeset_revision=constants_phaeo.ORGANISM_SYNC_TOOL_CHANGESET_REVISION,
instance=self.instance)

Arthur Le Bars
committed

Arthur Le Bars
committed
logging.info("Success: individual tools versions and changesets validated")

Loraine Gueguen
committed
def add_analysis(self, name, programversion, sourcename):
add_analysis_tool_dataset = utilities_bioblend.run_tool_and_download_single_output_dataset(

Loraine Gueguen
committed
instance=self.instance,
tool_id=constants_phaeo.ADD_ANALYSIS_TOOL_ID,

Loraine Gueguen
committed
history_id=self.history_id,
tool_inputs={"name": name,
"program": constants_phaeo.ADD_ANALYSIS_TOOL_PARAM_PROGRAM,

Loraine Gueguen
committed
"programversion": programversion,
"sourcename": sourcename,
"date_executed": constants_phaeo.ADD_ANALYSIS_TOOL_PARAM_DATE})

Loraine Gueguen
committed
analysis_dict = json.loads(add_analysis_tool_dataset)
analysis_id = str(analysis_dict["analysis_id"])

Loraine Gueguen
committed
return analysis_id

Arthur Le Bars
committed

Loraine Gueguen
committed
def sync_analysis(self, analysis_id):

Arthur Le Bars
committed

Loraine Gueguen
committed
time.sleep(60)
utilities_bioblend.run_tool(

Loraine Gueguen
committed
instance=self.instance,
tool_id=constants_phaeo.ANALYSIS_SYNC_TOOL_ID,

Loraine Gueguen
committed
history_id=self.history_id,
tool_inputs={"analysis_id": analysis_id})

Arthur Le Bars
committed

Loraine Gueguen
committed
def add_organism_and_sync(self):
get_organisms_tool_dataset = utilities_bioblend.run_tool_and_download_single_output_dataset(

Loraine Gueguen
committed
instance=self.instance,
tool_id=constants_phaeo.GET_ORGANISMS_TOOL_ID,

Arthur Le Bars
committed
history_id=self.history_id,

Loraine Gueguen
committed
tool_inputs={},
time_sleep=10
)
organisms_dict_list = json.loads(get_organisms_tool_dataset) # Turn the dataset into a list for parsing
# Look up list of outputs (dictionaries)

Loraine Gueguen
committed
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
add_organism_tool_dataset = utilities_bioblend.run_tool_and_download_single_output_dataset(

Loraine Gueguen
committed
instance=self.instance,
tool_id=constants_phaeo.ADD_ORGANISM_TOOL_ID,

Loraine Gueguen
committed
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

Arthur Le Bars
committed
# Synchronize newly added organism in Tripal

Arthur Le Bars
committed
logging.info("Synchronizing organism %s in Tripal" % self.full_name)
time.sleep(60)
utilities_bioblend.run_tool(

Loraine Gueguen
committed
instance=self.instance,
tool_id=constants_phaeo.ORGANISM_SYNC_TOOL_ID,

Loraine Gueguen
committed
history_id=self.history_id,
tool_inputs={"organism_id": org_id})

Arthur Le Bars
committed

Loraine Gueguen
committed
return org_id

Loraine Gueguen
committed
def get_analyses(self):
get_analyses_tool_dataset = utilities_bioblend.run_tool_and_download_single_output_dataset(

Loraine Gueguen
committed
instance=self.instance,
tool_id=constants_phaeo.GET_ANALYSES_TOOL_ID,
history_id=self.history_id,

Loraine Gueguen
committed
tool_inputs={},
time_sleep=10
)
analyses_dict_list = json.loads(get_analyses_tool_dataset)
return analyses_dict_list

Loraine Gueguen
committed
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)
"""

Loraine Gueguen
committed
analysis_id = None
# Look up list of outputs (dictionaries)

Loraine Gueguen
committed
for analyses_dict in analyses_dict_list:
if analyses_dict["name"] == analysis_name:
analysis_id = str(analyses_dict["analysis_id"])

Loraine Gueguen
committed
if analysis_id is None:
analysis_id = self.add_analysis(
name=analysis_name,
programversion=analysis_programversion,
sourcename=analysis_sourcename
)

Loraine Gueguen
committed
# Synchronize analysis in Tripal
logging.info("Synchronizing analysis %s in Tripal" % analysis_name)
self.sync_analysis(analysis_id=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:
"""

Loraine Gueguen
committed
self.set_galaxy_instance()
self.set_history()
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
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:

Loraine Gueguen
committed
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

Arthur Le Bars
committed
# 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})
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
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"])

Arthur Le Bars
committed
# 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})

Arthur Le Bars
committed
# 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:

Loraine Gueguen
committed
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,

Arthur Le Bars
committed
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:

Loraine Gueguen
committed
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,

Arthur Le Bars
committed
history_id=self.history_id,
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
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})

Arthur Le Bars
committed
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

Arthur Le Bars
committed
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)

Arthur Le Bars
committed
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)

Arthur Le Bars
committed
invocation_id = invocations[1]["id"] # Most recent invocation
invocation_report = self.instance.invocations.get_invocation_report(invocation_id=invocation_id)
logging.debug(invocation_report)

Arthur Le Bars
committed
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
genome_ldda_id = None
transcripts_ldda_id = None
proteins_ldda_id = None
gff_ldda_id = None
interpro_ldda_id = None
blastp_ldda_id = None
blastx_ldda_id = None
genome_hda_id = None
gff_hda_id = None
transcripts_hda_id = None
proteins_hda_id = None
blastp_hda_id = None
blastx_hda_id = None
interproscan_hda_id = None
folder_dict_list = self.instance.libraries.get_folders(library_id=str(self.library_id))

Loraine Gueguen
committed
folders_id_dict = {}

Arthur Le Bars
committed
# Loop over the folders in the library and map folders names to their IDs

Loraine Gueguen
committed
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

Loraine Gueguen
committed
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 value in sub_folder_content.values():
for e in value:
if e["name"].endswith(self.genome_filename):
genome_ldda_id = e["ldda_id"]

Arthur Le Bars
committed

Loraine Gueguen
committed
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 value in sub_folder_content.values():
for e in value:
ldda_name = e["name"]
ldda_id = e["ldda_id"]
if ldda_name.endswith(self.transcripts_filename):
transcripts_ldda_id = ldda_id
elif ldda_name.endswith(self.proteins_filename):
proteins_ldda_id = ldda_id
elif ldda_name.endswith(self.gff_filename):
gff_ldda_id = ldda_id
elif ldda_name.endswith(self.interpro_filename):
interpro_ldda_id = ldda_id
elif ldda_name.endswith(self.blastp_filename):
blastp_ldda_id = ldda_id
elif ldda_name.endswith(self.blastx_filename):
blastx_ldda_id = ldda_id
hda_list = self.instance.datasets.get_datasets(self.history_id)
# Finding datasets in history (matching datasets names)
for hda in hda_list:
hda_name = hda["name"]
hda_id = hda["id"]
if hda_name == self.genome_filename:
genome_hda_id = hda_id
if hda_name == self.gff_filename:
gff_hda_id = hda_id
if hda_name == self.transcripts_filename:
transcripts_hda_id = hda_id
if hda_name == self.proteins_filename :
proteins_hda_id = hda_id
if hda_name == self.blastp_filename:
blastp_hda_id = hda_id
if hda_name == self.blastx_filename:
blastx_hda_id = hda_id
if hda_name == self.interpro_filename:
interproscan_hda_id = hda_id
# Import each dataset into history if it is not imported
logging.debug("Uploading datasets into history %s" % self.history_id)

Arthur Le Bars
committed
genome_dataset_upload = self.instance.histories.upload_dataset_from_library(history_id=self.history_id, lib_dataset_id=genome_ldda_id)
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=gff_ldda_id)
gff_hda_id = gff_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=proteins_ldda_id)
proteins_hda_id = proteins_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=transcripts_ldda_id)
transcripts_hda_id = transcripts_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=interpro_ldda_id)
interproscan_hda_id = interproscan_dataset_upload["id"]
except Exception as exc:

Arthur Le Bars
committed
logging.debug("Interproscan file not found in library (history: {0})".format(self.history_id))
try:
blastp_dataset_upload = self.instance.histories.upload_dataset_from_library(history_id=self.history_id, lib_dataset_id=blastp_ldda_id)
except Exception as exc:
logging.debug("blastp file not found in library (history: {0})".format(self.history_id))
if blastx_hda_id is None:
try:
blastx_dataset_upload = self.instance.histories.upload_dataset_from_library(history_id=self.history_id, lib_dataset_id=blastx_ldda_id)
blastx_hda_id = blastx_dataset_upload["id"]
except Exception as exc:
logging.debug("blastp file not found in library (history: {0})".format(self.history_id))
self.genome_hda_id = genome_hda_id
self.gff_hda_id = gff_hda_id
self.transcripts_hda_id = transcripts_hda_id
self.proteins_hda_id = proteins_hda_id
self.blastp_hda_id = blastp_hda_id
self.blastx_hda_id = blastx_hda_id
self.interproscan_hda_id = interproscan_hda_id
def get_sp_workflow_param(sp_dict, main_dir, config, workflow_type):
"""
"""
run_workflow_for_current_organism = RunWorkflow(parameters_dictionary=sp_dict)
# Verifying the galaxy container is running
if not utilities_bioblend.check_galaxy_state(network_name=run_workflow_for_current_organism.genus_species,
script_dir=run_workflow_for_current_organism.script_dir):
logging.critical("The galaxy container for %s is not ready yet!" % run_workflow_for_current_organism.genus_species)
sys.exit()
else:
# 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

Loraine Gueguen
committed
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)
run_workflow_for_current_organism.instance = utilities_bioblend.get_galaxy_instance(
instance_url=run_workflow_for_current_organism.instance_url,
email=run_workflow_for_current_organism.config[constants.CONF_GALAXY_DEFAULT_ADMIN_EMAIL],
password=run_workflow_for_current_organism.config[constants.CONF_GALAXY_DEFAULT_ADMIN_PASSWORD],
)
history_id = utilities_bioblend.get_history(
instance=run_workflow_for_current_organism.instance,
history_name=run_workflow_for_current_organism.history_name)
run_workflow_for_current_organism.install_changesets_revisions_for_individual_tools()
if workflow_type == constants_phaeo.WF_LOAD_GFF_JB:

Loraine Gueguen
committed
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
)
run_workflow_for_current_organism.import_datasets_into_history()
# Create the StrainWorkflowParam object holding all attributes needed for the workflow
sp_wf_param = StrainWorkflowParam(
genus_species=run_workflow_for_current_organism.genus_species,
strain_sex=run_workflow_for_current_organism.strain_sex,
genus_uppercase = run_workflow_for_current_organism.genus_uppercase,
full_name=run_workflow_for_current_organism.full_name,
species_folder_name=run_workflow_for_current_organism.species_folder_name,
chado_species_name=run_workflow_for_current_organism.chado_species_name,
org_id=org_id,
genome_analysis_id=genome_analysis_id,
ogs_analysis_id=ogs_analysis_id,
genome_hda_id=run_workflow_for_current_organism.genome_hda_id,
gff_hda_id=run_workflow_for_current_organism.gff_hda_id,
transcripts_hda_id=run_workflow_for_current_organism.transcripts_hda_id,
proteins_hda_id=run_workflow_for_current_organism.proteins_hda_id,
blastp_hda_id=run_workflow_for_current_organism.blastp_hda_id,
blastx_hda_id=run_workflow_for_current_organism.blastx_hda_id,
interproscan_hda_id=run_workflow_for_current_organism.interproscan_hda_id,
instance=run_workflow_for_current_organism.instance
sp_wf_param.check_param_for_workflow_load_fasta_gff_jbrowse()
if workflow_type == "blast":
ids = run_workflow_for_current_organism.add_organism_blastp_analysis()
org_id = ids["org_id"]
blastp_analysis_id = ids["blastp_analysis_id"]
run_workflow_for_current_organism.import_datasets_into_history()
# Create the StrainWorkflowParam object holding all attributes needed for the workflow
sp_wf_param = StrainWorkflowParam(
genus_species=run_workflow_for_current_organism.genus_species,
strain_sex=run_workflow_for_current_organism.strain_sex,
genus_uppercase = run_workflow_for_current_organism.genus_uppercase,
full_name=run_workflow_for_current_organism.full_name,
species_folder_name=run_workflow_for_current_organism.species_folder_name,
chado_species_name=run_workflow_for_current_organism.chado_species_name,
blastp_analysis_id=blastp_analysis_id,
genome_hda_id=run_workflow_for_current_organism.genome_hda_id,
gff_hda_id=run_workflow_for_current_organism.gff_hda_id,
transcripts_hda_id=run_workflow_for_current_organism.transcripts_hda_id,
proteins_hda_id=run_workflow_for_current_organism.proteins_hda_id,
blastp_hda_id=run_workflow_for_current_organism.blastp_hda_id,
blastx_hda_id=run_workflow_for_current_organism.blastx_hda_id,
interproscan_hda_id=run_workflow_for_current_organism.interproscan_hda_id,
instance=run_workflow_for_current_organism.instance
sp_wf_param.check_param_for_workflow_blastp()
if workflow_type == "interpro":
ids = run_workflow_for_current_organism.add_organism_interproscan_analysis()
org_id = ids["org_id"]
interpro_analysis_id = ids["interpro_analysis_id"]
run_workflow_for_current_organism.import_datasets_into_history()
# Create the StrainWorkflowParam object holding all attributes needed for the workflow
sp_wf_param = StrainWorkflowParam(
genus_species=run_workflow_for_current_organism.genus_species,
strain_sex=run_workflow_for_current_organism.strain_sex,
genus_uppercase = run_workflow_for_current_organism.genus_uppercase,
full_name=run_workflow_for_current_organism.full_name,
species_folder_name=run_workflow_for_current_organism.species_folder_name,
chado_species_name=run_workflow_for_current_organism.chado_species_name,
org_id=org_id,
interpro_analysis_id=interpro_analysis_id,
genome_hda_id=run_workflow_for_current_organism.genome_hda_id,
gff_hda_id=run_workflow_for_current_organism.gff_hda_id,
transcripts_hda_id=run_workflow_for_current_organism.transcripts_hda_id,
proteins_hda_id=run_workflow_for_current_organism.proteins_hda_id,
blastp_hda_id=run_workflow_for_current_organism.blastp_hda_id,
blastx_hda_id=run_workflow_for_current_organism.blastx_hda_id,
interproscan_hda_id=run_workflow_for_current_organism.interproscan_hda_id,
instance=run_workflow_for_current_organism.instance
sp_wf_param.check_param_for_workflow_interpro()
return sp_wf_param
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 step in workflow_dict["steps"].values():
if step["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")
utilities_bioblend.install_repository_revision(tool_id=step["tool_id"],
version=step["tool_version"],
changeset_revision=step["tool_shed_repository"]["changeset_revision"],
instance=instance)
logging.info("Tools versions and changeset_revisions from workflow validated")

Arthur Le Bars
committed
if __name__ == "__main__":

Loraine Gueguen
committed
parser = argparse.ArgumentParser(description="Run Galaxy workflows, specific to Phaeoexplorer data")

Arthur Le Bars
committed
parser.add_argument("input",
type=str,
help="Input file (yml)")
parser.add_argument("-v", "--verbose",
help="Increase output verbosity",

Arthur Le Bars
committed
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")

Arthur Le Bars
committed

Arthur Le Bars
committed
parser.add_argument("--workflow", "-w",
type=str,
help="Worfklow to run. Available options: load_fasta_gff_jbrowse, blast, interpro")

Arthur Le Bars
committed

Arthur Le Bars
committed
args = parser.parse_args()

Arthur Le Bars
committed
if args.verbose:
logging.basicConfig(level=logging.DEBUG)

Arthur Le Bars
committed
else:
logging.basicConfig(level=logging.INFO)

Arthur Le Bars
committed

Arthur Le Bars
committed
# 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)

Loraine Gueguen
committed
main_dir = None

Arthur Le Bars
committed
if not args.main_directory:

Loraine Gueguen
committed
main_dir = os.getcwd()

Arthur Le Bars
committed
else:

Loraine Gueguen
committed
main_dir = os.path.abspath(args.main_directory)

Arthur Le Bars
committed
sp_dict_list = utilities.parse_input(args.input)

Arthur Le Bars
committed
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 constants_phaeo.WORKFLOW_VALID_TYPES:

Loraine Gueguen
committed
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 == constants_phaeo.WF_LOAD_GFF_JB:

Arthur Le Bars
committed
# Add and retrieve all analyses/organisms for the current input species and add their IDs to the input dictionary
sp_wf_param = get_sp_workflow_param(

Loraine Gueguen
committed
sp_dict,
main_dir=main_dir,
config=config,
workflow_type=constants_phaeo.WF_LOAD_GFF_JB)
current_sp_genus_species = sp_wf_param.genus_species
current_sp_strain_sex = sp_wf_param.strain_sex
# 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_genus_species in all_sp_workflow_dict.keys():
all_sp_workflow_dict[current_sp_genus_species] = {current_sp_strain_sex: sp_wf_param}
if not current_sp_strain_sex in all_sp_workflow_dict[current_sp_genus_species].keys():
all_sp_workflow_dict[current_sp_genus_species][current_sp_strain_sex] = sp_wf_param
else:
logging.error("Duplicate organism with 'genus_species' = '{0}' and 'strain_sex' = '{1}'".format(current_sp_genus_species, current_sp_strain_sex))
for species, strains in all_sp_workflow_dict.items():
strains_list = list(strains.keys())
strains_count = len(strains_list)
logging.info("Input species %s: 1 strain detected in input dictionary" % species)
strain_sex = list(strains.keys())[0]
sp_wf_param = strains[strain_sex]
workflow_path = os.path.join(os.path.abspath(script_dir), constants_phaeo.WORKFLOWS_PATH, constants_phaeo.WF_LOAD_GFF_JB_1_ORG_FILE)
# Check if the versions of tools specified in the workflow are installed in galaxy
install_changesets_revisions_from_workflow(workflow_path=workflow_path, instance=sp_wf_param.instance)
# 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[constants_phaeo.WF_LOAD_GFF_JB_1_ORG_INPUT_GENOME] = {}
workflow_parameters[constants_phaeo.WF_LOAD_GFF_JB_1_ORG_INPUT_GFF] = {}
workflow_parameters[constants_phaeo.WF_LOAD_GFF_JB_1_ORG_INPUT_PROTEINS] = {}
workflow_parameters[constants_phaeo.WF_LOAD_GFF_JB_1_ORG_STEP_LOAD_FASTA] = {
"organism": sp_wf_param.org_id,
"analysis_id": sp_wf_param.genome_analysis_id,
workflow_parameters[constants_phaeo.WF_LOAD_GFF_JB_1_ORG_STEP_JBROWSE] = {}
workflow_parameters[constants_phaeo.WF_LOAD_GFF_JB_1_ORG_STEP_LOAD_GFF] = {
"organism": sp_wf_param.org_id,
"analysis_id": sp_wf_param.ogs_analysis_id}
workflow_parameters[constants_phaeo.WF_LOAD_GFF_JB_1_ORG_STEP_FEATURE_SYNC] = {
"organism_id": sp_wf_param.org_id}
workflow_parameters[constants_phaeo.WF_LOAD_GFF_JB_1_ORG_STEP_POPULATE_VIEWS] = {}
workflow_parameters[constants_phaeo.WF_LOAD_GFF_JB_1_ORG_STEP_INDEX] = {}
# Set datamap (mapping of input files in the workflow)
datamap = {}
datamap[constants_phaeo.WF_LOAD_GFF_JB_1_ORG_INPUT_GENOME] = {"src": "hda", "id": sp_wf_param.genome_hda_id}
datamap[constants_phaeo.WF_LOAD_GFF_JB_1_ORG_INPUT_GFF] = {"src": "hda", "id": sp_wf_param.gff_hda_id}
datamap[constants_phaeo.WF_LOAD_GFF_JB_1_ORG_INPUT_PROTEINS] = {"src": "hda", "id": sp_wf_param.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)
if constants.CONF_JBROWSE_MENU_URL not in config.keys():
# default
root_url = "https://{0}".format(config[constants.CONF_ALL_HOSTNAME])
root_url = config[constants.CONF_JBROWSE_MENU_URL]
species_strain_sex = sp_wf_param.chado_species_name.replace(" ", "-")
jbrowse_menu_url = "{root_url}/sp/{genus_sp}/feature/{Genus}/{species_strain_sex}/mRNA/{id}".format(
root_url=root_url,
genus_sp=sp_wf_param.genus_species,
Genus=sp_wf_param.genus_uppercase,
species_strain_sex=species_strain_sex,
id="{id}")
jbrowse_tool_state = workflow_dict["steps"][constants_phaeo.WF_LOAD_GFF_JB_1_ORG_STEP_JBROWSE]["tool_state"]
jbrowse_tool_state = jbrowse_tool_state.replace("__MENU_URL_ORG__", jbrowse_menu_url)
jb_to_container_tool_state = workflow_dict["steps"][constants_phaeo.WF_LOAD_GFF_JB_1_ORG_STEP_JB_TO_CONTAINER]["tool_state"]
jb_to_container_tool_state = jb_to_container_tool_state\
.replace("__DISPLAY_NAME_ORG__", sp_wf_param.full_name)\
.replace("__UNIQUE_ID_ORG__", sp_wf_param.species_folder_name)
# Import the workflow in galaxy as a dict
sp_wf_param.instance.workflows.import_workflow_dict(workflow_dict=workflow_dict)
workflow_attributes = sp_wf_param.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 = sp_wf_param.instance.workflows.show_workflow(workflow_id=workflow_id)
except bioblend.ConnectionError:
logging.warning("Error finding workflow %s" % workflow_name)
# Finally, invoke the workflow along with its datamap, parameters and the history in which to invoke it
sp_wf_param.instance.workflows.invoke_workflow(
history_id=sp_wf_param.history_id,
params=workflow_parameters,
inputs=datamap,
allow_tool_state_corrections=True)
logging.info("Successfully imported and invoked workflow {0}, check the galaxy instance for the jobs state".format(workflow_name))
logging.info("Input organism %s: 2 species detected in input dictionary" % species)
strain_sex_org1 = strains_list[0]
strain_sex_org2 = strains_list[1]