annotate env/lib/python3.9/site-packages/planemo/galaxy/activity.py @ 0:4f3585e2f14b draft default tip

"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"

1 """Module provides generic interface to running Galaxy tools and workflows."""
2 
3 import json
4 import os
5 import sys
6 import tempfile
7 import time
8 import traceback
9 
10 import bioblend
11 from bioblend.util import attach_file
12 from galaxy.tool_util.client.staging import (
13     StagingInterace,
14 )
15 from galaxy.tool_util.cwl.util import (
16     invocation_to_output,
17     output_properties,
18     output_to_cwl_json,
19     tool_response_to_output,
20 )
21 from galaxy.tool_util.parser import get_tool_source
22 from galaxy.tool_util.verify.interactor import galaxy_requests_post
23 from galaxy.util import (
24     safe_makedirs,
25     unicodify,
26 )
27 from requests.exceptions import RequestException
28 from six.moves.urllib.parse import urljoin
29 
30 from planemo.galaxy.api import summarize_history
31 from planemo.io import wait_on
32 from planemo.runnable import (
33     ErrorRunResponse,
34     get_outputs,
35     RunnableType,
36     SuccessfulRunResponse,
37 )
38 
39 DEFAULT_HISTORY_NAME = "CWL Target History"
40 ERR_NO_SUCH_TOOL = ("Failed to find tool with ID [%s] in Galaxy - cannot execute job. "
41                     "You may need to enable verbose logging and determine why the tool did not load. [%s]")
42 
43 
44 def execute(ctx, config, runnable, job_path, **kwds):
45     """Execute a Galaxy activity."""
46     try:
47         return _execute(ctx, config, runnable, job_path, **kwds)
48     except Exception as e:
49         if ctx.verbose:
50             ctx.vlog("Failed to execute Galaxy activity, throwing ErrorRunResponse")
51             traceback.print_exc(file=sys.stdout)
52         return ErrorRunResponse(unicodify(e))
53 
54 
55 def _verified_tool_id(runnable, user_gi):
56     tool_id = _tool_id(runnable.path)
57     try:
58         user_gi.tools.show_tool(tool_id)
59     except Exception as e:
60         raise Exception(ERR_NO_SUCH_TOOL % (tool_id, e))
61     return tool_id
62 
63 
64 def _inputs_representation(runnable):
65     if runnable.type == RunnableType.cwl_tool:
66         inputs_representation = "cwl"
67     else:
68         inputs_representation = "galaxy"
69     return inputs_representation
70 
71 
72 def log_contents_str(config):
73     if hasattr(config, "log_contents"):
74         return config.log_contents
75     else:
76         return "No log for this engine type."
77 
78 
79 class PlanemoStagingInterface(StagingInterace):
80 
81     def __init__(self, ctx, runnable, user_gi, version_major):
82         self._ctx = ctx
83         self._user_gi = user_gi
84         self._runnable = runnable
85         self._version_major = version_major
86 
87     def _post(self, api_path, payload, files_attached=False):
88         params = dict(key=self._user_gi.key)
89         url = urljoin(self._user_gi.url, "api/" + api_path)
90         return galaxy_requests_post(url, data=payload, params=params, as_json=True).json()
91 
92     def _attach_file(self, path):
93         return attach_file(path)
94 
95     def _handle_job(self, job_response):
96         job_id = job_response["id"]
97         _wait_for_job(self._user_gi, job_id)
98 
99     @property
00     def use_fetch_api(self):
01         # hack around this not working for galaxy_tools - why is that :(
02         return self._runnable.type != RunnableType.galaxy_tool and self._version_major >= "20.09"
03 
04     # extension point for planemo to override logging
05     def _log(self, message):
06         self._ctx.vlog(message)
07 
08 
09 def _execute(ctx, config, runnable, job_path, **kwds):
10     user_gi = config.user_gi
11     admin_gi = config.gi
12 
13     history_id = _history_id(user_gi, **kwds)
14 
15     try:
16         job_dict, _ = stage_in(ctx, runnable, config, user_gi, history_id, job_path, **kwds)
17     except Exception:
18         ctx.vlog("Problem with staging in data for Galaxy activities...")
19         raise
20     if runnable.type in [RunnableType.galaxy_tool, RunnableType.cwl_tool]:
21         response_class = GalaxyToolRunResponse
22         tool_id = _verified_tool_id(runnable, user_gi)
23         inputs_representation = _inputs_representation(runnable)
24         run_tool_payload = dict(
25             history_id=history_id,
26             tool_id=tool_id,
27             inputs=job_dict,
28             inputs_representation=inputs_representation,
29         )
30         ctx.vlog("Post to Galaxy tool API with payload [%s]" % run_tool_payload)
31         tool_run_response = user_gi.tools._post(run_tool_payload)
32 
33         job = tool_run_response["jobs"][0]
34         job_id = job["id"]
35         try:
36             final_state = _wait_for_job(user_gi, job_id)
37         except Exception:
38             summarize_history(ctx, user_gi, history_id)
39             raise
40         if final_state != "ok":
41             msg = "Failed to run CWL tool job final job state is [%s]." % final_state
42             summarize_history(ctx, user_gi, history_id)
43             raise Exception(msg)
44 
45         ctx.vlog("Final job state was ok, fetching details for job [%s]" % job_id)
46         job_info = admin_gi.jobs.show_job(job_id)
47         response_kwds = {
48             'job_info': job_info,
49             'api_run_response': tool_run_response,
50         }
51         if ctx.verbose:
52             summarize_history(ctx, user_gi, history_id)
53     elif runnable.type in [RunnableType.galaxy_workflow, RunnableType.cwl_workflow]:
54         response_class = GalaxyWorkflowRunResponse
55         workflow_id = config.workflow_id_for_runnable(runnable)
56         ctx.vlog("Found Galaxy workflow ID [%s] for URI [%s]" % (workflow_id, runnable.uri))
57         # TODO: Use the following when BioBlend 0.14 is released
58         # invocation = user_gi.worklfows.invoke_workflow(
59         #    workflow_id,
60         #    inputs=job_dict,
61         #    history_id=history_id,
62         #    allow_tool_state_corrections=True,
63         #    inputs_by="name",
64         # )
65         payload = dict(
66             workflow_id=workflow_id,
67             history_id=history_id,
68             inputs=job_dict,
69             inputs_by="name",
70             allow_tool_state_corrections=True,
71         )
72         invocations_url = "%s/workflows/%s/invocations" % (
73             user_gi.url,
74             workflow_id,
75         )
76         invocation = user_gi.workflows._post(payload, url=invocations_url)
77         invocation_id = invocation["id"]
78         ctx.vlog("Waiting for invocation [%s]" % invocation_id)
79         polling_backoff = kwds.get("polling_backoff", 0)
80         final_invocation_state = 'new'
81         error_message = ""
82         try:
83             final_invocation_state = _wait_for_invocation(ctx, user_gi, history_id, workflow_id, invocation_id, polling_backoff)
84             assert final_invocation_state == 'scheduled'
85         except Exception:
86             ctx.vlog("Problem waiting on invocation...")
87             summarize_history(ctx, user_gi, history_id)
88             error_message = "Final invocation state is [%s]" % final_invocation_state
89         ctx.vlog("Final invocation state is [%s]" % final_invocation_state)
90         final_state = _wait_for_history(ctx, user_gi, history_id, polling_backoff)
91         if final_state != "ok":
92             msg = "Failed to run workflow final history state is [%s]." % final_state
93             error_message = msg if not error_message else "%s. %s" % (error_message, msg)
94             ctx.vlog(msg)
95             summarize_history(ctx, user_gi, history_id)
96         else:
97             ctx.vlog("Final history state is 'ok'")
98         response_kwds = {
99             'workflow_id': workflow_id,
00             'invocation_id': invocation_id,
01             'history_state': final_state,
02             'invocation_state': final_invocation_state,
03             'error_message': error_message,
04         }
05     else:
06         raise NotImplementedError()
07 
08     run_response = response_class(
09         ctx=ctx,
10         runnable=runnable,
11         user_gi=user_gi,
12         history_id=history_id,
13         log=log_contents_str(config),
14         **response_kwds
15     )
16     output_directory = kwds.get("output_directory", None)
17     ctx.vlog("collecting outputs from run...")
18     run_response.collect_outputs(ctx, output_directory)
19     ctx.vlog("collecting outputs complete")
20     return run_response
21 
22 
23 def stage_in(ctx, runnable, config, user_gi, history_id, job_path, **kwds):  # noqa C901
24     # only upload objects as files/collections for CWL workflows...
25     tool_or_workflow = "tool" if runnable.type != RunnableType.cwl_workflow else "workflow"
26     to_posix_lines = runnable.type.is_galaxy_artifact
27     job_dict, datasets = PlanemoStagingInterface(ctx, runnable, user_gi, config.version_major).stage(
28         tool_or_workflow,
29         history_id=history_id,
30         job_path=job_path,
31         use_path_paste=config.use_path_paste,
32         to_posix_lines=to_posix_lines,
33     )
34 
35     if datasets:
36         ctx.vlog("uploaded datasets [%s] for activity, checking history state" % datasets)
37         final_state = _wait_for_history(ctx, user_gi, history_id)
38 
39         for (dataset, path) in datasets:
40             dataset_details = user_gi.histories.show_dataset(
41                 history_id,
42                 dataset["id"],
43             )
44             ctx.vlog("Uploaded dataset for path [%s] with metadata [%s]" % (path, dataset_details))
45     else:
46         # Mark uploads as ok because nothing to do.
47         final_state = "ok"
48 
49     ctx.vlog("final state is %s" % final_state)
50     if final_state != "ok":
51         msg = "Failed to run job final job state is [%s]." % final_state
52         summarize_history(ctx, user_gi, history_id)
53         raise Exception(msg)
54 
55     return job_dict, datasets
56 
57 
58 def _file_path_to_name(file_path):
59     if file_path is not None:
60         name = os.path.basename(file_path)
61     else:
62         name = "defaultname"
63     return name
64 
65 
66 class GalaxyBaseRunResponse(SuccessfulRunResponse):
67 
68     def __init__(
69         self,
70         ctx,
71         runnable,
72         user_gi,
73         history_id,
74         log,
75     ):
76         self._ctx = ctx
77         self._runnable = runnable
78         self._user_gi = user_gi
79         self._history_id = history_id
80         self._log = log
81 
82         self._job_info = None
83 
84         self._outputs_dict = None
85 
86     def to_galaxy_output(self, output):
87         """Convert runnable output to a GalaxyOutput object.
88 
89         Subclasses for workflow and tool execution override this.
90         """
91         raise NotImplementedError()
92 
93     def _get_extra_files(self, dataset_details):
94         extra_files_url = "%s/histories/%s/contents/%s/extra_files" % (
95             self._user_gi.url, self._history_id, dataset_details["id"]
96         )
97         extra_files = self._user_gi.jobs._get(url=extra_files_url)
98         return extra_files
99 
00     def _get_metadata(self, history_content_type, content_id):
01         if history_content_type == "dataset":
02             return self._user_gi.histories.show_dataset(
03                 self._history_id,
04                 content_id,
05             )
06         elif history_content_type == "dataset_collection":
07             return self._user_gi.histories.show_dataset_collection(
08                 self._history_id,
09                 content_id,
10             )
11         else:
12             raise Exception("Unknown history content type encountered [%s]" % history_content_type)
13 
14     def collect_outputs(self, ctx, output_directory):
15         assert self._outputs_dict is None, "collect_outputs pre-condition violated"
16 
17         outputs_dict = {}
18         if not output_directory:
19             # TODO: rather than creating a directory just use
20             # Galaxy paths if they are available in this
21             # configuration.
22             output_directory = tempfile.mkdtemp()
23 
24         def get_dataset(dataset_details, filename=None):
25             parent_basename = dataset_details.get("cwl_file_name")
26             if not parent_basename:
27                 parent_basename = dataset_details.get("name")
28             file_ext = dataset_details["file_ext"]
29             if file_ext == "directory":
30                 # TODO: rename output_directory to outputs_directory because we can have output directories
31                 # and this is confusing...
32                 the_output_directory = os.path.join(output_directory, parent_basename)
33                 safe_makedirs(the_output_directory)
34                 destination = self.download_output_to(dataset_details, the_output_directory, filename=filename)
35             else:
36                 destination = self.download_output_to(dataset_details, output_directory, filename=filename)
37             if filename is None:
38                 basename = parent_basename
39             else:
40                 basename = os.path.basename(filename)
41 
42             return {"path": destination, "basename": basename}
43 
44         ctx.vlog("collecting outputs to directory %s" % output_directory)
45 
46         for runnable_output in get_outputs(self._runnable, gi=self._user_gi):
47             output_id = runnable_output.get_id()
48             if not output_id:
49                 ctx.vlog("Workflow output identified without an ID (label), skipping")
50                 continue
51             output_dict_value = None
52             is_cwl = self._runnable.type in [RunnableType.cwl_workflow, RunnableType.cwl_tool]
53             output_src = self.output_src(runnable_output)
54             if not is_cwl and output_src["src"] == "hda":
55                 output_dataset_id = output_src["id"]
56                 dataset = self._get_metadata("dataset", output_dataset_id)
57                 dataset_dict = get_dataset(dataset)
58                 ctx.vlog("populated destination [%s]" % dataset_dict["path"])
59 
60                 if dataset["file_ext"] == "expression.json":
61                     with open(dataset_dict["path"], "r") as f:
62                         output_dict_value = json.load(f)
63                 else:
64                     output_dict_value = output_properties(**dataset_dict)
65             else:
66                 output_dataset_id = output_src["id"]
67                 galaxy_output = self.to_galaxy_output(runnable_output)
68                 cwl_output = output_to_cwl_json(
69                     galaxy_output,
70                     self._get_metadata,
71                     get_dataset,
72                     self._get_extra_files,
73                     pseduo_location=True,
74                 )
75                 if is_cwl:
76                     output_dict_value = cwl_output
77                 else:
78 
79                     def attach_file_properties(collection, cwl_output):
80                         elements = collection["elements"]
81                         assert len(elements) == len(cwl_output)
82                         for element, cwl_output_element in zip(elements, cwl_output):
83                             element["_output_object"] = cwl_output_element
84                             if isinstance(cwl_output_element, list):
85                                 assert "elements" in element["object"]
86                                 attach_file_properties(element["object"], cwl_output_element)
87 
88                     output_metadata = self._get_metadata("dataset_collection", output_dataset_id)
89                     attach_file_properties(output_metadata, cwl_output)
90                     output_dict_value = output_metadata
91 
92             outputs_dict[output_id] = output_dict_value
93 
94         self._outputs_dict = outputs_dict
95         ctx.vlog("collected outputs [%s]" % self._outputs_dict)
96 
97     @property
98     def log(self):
99         return self._log
00 
01     @property
02     def job_info(self):
03         if self._job_info is not None:
04             return dict(
05                 stdout=self._job_info["stdout"],
06                 stderr=self._job_info["stderr"],
07                 command_line=self._job_info["command_line"],
08             )
09         return None
10 
11     @property
12     def invocation_details(self):
13         return None
14 
15     @property
16     def outputs_dict(self):
17         return self._outputs_dict
18 
19     def download_output_to(self, dataset_details, output_directory, filename=None):
20         if filename is None:
21             local_filename = dataset_details.get("cwl_file_name") or dataset_details.get("name")
22         else:
23             local_filename = filename
24         destination = os.path.join(output_directory, local_filename)
25         self._history_content_download(
26             self._history_id,
27             dataset_details["id"],
28             to_path=destination,
29             filename=filename,
30         )
31         return destination
32 
33     def _history_content_download(self, history_id, dataset_id, to_path, filename=None):
34         user_gi = self._user_gi
35         url = user_gi.url + "/histories/%s/contents/%s/display" % (history_id, dataset_id)
36 
37         data = {}
38         if filename:
39             data["filename"] = filename
40 
41         r = user_gi.make_get_request(url, params=data, stream=True, timeout=user_gi.timeout)
42         r.raise_for_status()
43 
44         with open(to_path, 'wb') as fp:
45             for chunk in r.iter_content(chunk_size=bioblend.CHUNK_SIZE):
46                 if chunk:
47                     fp.write(chunk)
48 
49 
50 class GalaxyToolRunResponse(GalaxyBaseRunResponse):
51 
52     def __init__(
53         self,
54         ctx,
55         runnable,
56         user_gi,
57         history_id,
58         log,
59         job_info,
60         api_run_response,
61     ):
62         super(GalaxyToolRunResponse, self).__init__(
63             ctx=ctx,
64             runnable=runnable,
65             user_gi=user_gi,
66             history_id=history_id,
67             log=log,
68         )
69         self._job_info = job_info
70         self.api_run_response = api_run_response
71 
72     def is_collection(self, output):
73         # TODO: Make this more rigorous - search both output and output
74         # collections - throw an exception if not found in either place instead
75         # of just assuming all non-datasets are collections.
76         return self.output_src(output)["src"] == "hdca"
77 
78     def to_galaxy_output(self, runnable_output):
79         output_id = runnable_output.get_id()
80         return tool_response_to_output(self.api_run_response, self._history_id, output_id)
81 
82     def output_src(self, output):
83         outputs = self.api_run_response["outputs"]
84         output_collections = self.api_run_response["output_collections"]
85         output_id = output.get_id()
86         output_src = None
87         self._ctx.vlog("Looking for id [%s] in outputs [%s]" % (output_id, outputs))
88         for output in outputs:
89             if output["output_name"] == output_id:
90                 output_src = {"src": "hda", "id": output["id"]}
91         for output_collection in output_collections:
92             if output_collection["output_name"] == output_id:
93                 output_src = {"src": "hdca", "id": output_collection["id"]}
94         return output_src
95 
96 
97 class GalaxyWorkflowRunResponse(GalaxyBaseRunResponse):
98 
99     def __init__(
00         self,
01         ctx,
02         runnable,
03         user_gi,
04         history_id,
05         log,
06         workflow_id,
07         invocation_id,
08         history_state='ok',
09         invocation_state='ok',
10         error_message=None,
11     ):
12         super(GalaxyWorkflowRunResponse, self).__init__(
13             ctx=ctx,
14             runnable=runnable,
15             user_gi=user_gi,
16             history_id=history_id,
17             log=log,
18         )
19         self._workflow_id = workflow_id
20         self._invocation_id = invocation_id
21         self._invocation_details = {}
22         self._cached_invocation = None
23         self.history_state = history_state
24         self.invocation_state = invocation_state
25         self.error_message = error_message
26         self._invocation_details = self.collect_invocation_details(invocation_id)
27 
28     def to_galaxy_output(self, runnable_output):
29         output_id = runnable_output.get_id()
30         self._ctx.vlog("checking for output in invocation [%s]" % self._invocation)
31         return invocation_to_output(self._invocation, self._history_id, output_id)
32 
33     def output_src(self, output):
34         invocation = self._invocation
35         # Use newer workflow outputs API.
36 
37         output_name = output.get_id()
38         if output_name in invocation["outputs"]:
39             return invocation["outputs"][output.get_id()]
40         elif output_name in invocation["output_collections"]:
41             return invocation["output_collections"][output.get_id()]
42         else:
43             raise Exception("Failed to find output [%s] in invocation outputs [%s]" % (output_name, invocation["outputs"]))
44 
45     def collect_invocation_details(self, invocation_id=None):
46         gi = self._user_gi
47         invocation_steps = {}
48         invocation = self.get_invocation(invocation_id)
49         for step in invocation['steps']:
50             step_label_or_index = "{}. {}".format(step['order_index'], step['workflow_step_label'] or 'Unnamed step')
51             workflow_step = gi.invocations.show_invocation_step(self._invocation_id, step['id'])
52             workflow_step['subworkflow'] = None
53             subworkflow_invocation_id = workflow_step.get('subworkflow_invocation_id')
54             if subworkflow_invocation_id:
55                 workflow_step['subworkflow'] = self.collect_invocation_details(subworkflow_invocation_id)
56             workflow_step_job_details = [self._user_gi.jobs.show_job(j['id'], full_details=True) for j in workflow_step['jobs']]
57             workflow_step['jobs'] = workflow_step_job_details
58             invocation_steps[step_label_or_index] = workflow_step
59         return invocation_steps
60 
61     @property
62     def invocation_details(self):
63         return self._invocation_details
64 
65     def get_invocation(self, invocation_id):
66         return self._user_gi.invocations.show_invocation(invocation_id)
67 
68     @property
69     def _invocation(self):
70         if self._cached_invocation is None:
71             self._cached_invocation = self.get_invocation(self._invocation_id)
72         return self._cached_invocation
73 
74     @property
75     def was_successful(self):
76         return self.history_state == 'ok' and self.invocation_state == 'scheduled'
77 
78 
79 def _tool_id(tool_path):
80     tool_source = get_tool_source(tool_path)
81     return tool_source.parse_id()
82 
83 
84 def _history_id(gi, **kwds):
85     history_id = kwds.get("history_id", None)
86     if history_id is None:
87         history_name = kwds.get("history_name", DEFAULT_HISTORY_NAME) or DEFAULT_HISTORY_NAME
88         history_id = gi.histories.create_history(history_name)["id"]
89     return history_id
90 
91 
92 def get_dict_from_workflow(gi, workflow_id):
93     return gi.workflows.export_workflow_dict(workflow_id)
94 
95 
96 def _wait_for_invocation(ctx, gi, history_id, workflow_id, invocation_id, polling_backoff=0):
97 
98     def state_func():
99         return _retry_on_timeouts(ctx, gi, lambda gi: gi.workflows.show_invocation(workflow_id, invocation_id))
00 
01     return _wait_on_state(state_func, polling_backoff)
02 
03 
04 def _retry_on_timeouts(ctx, gi, f):
05     gi.timeout = 60
06     try_count = 5
07     try:
08         for try_num in range(try_count):
09             start_time = time.time()
10             try:
11                 return f(gi)
12             except RequestException:
13                 end_time = time.time()
14                 if end_time - start_time > 45 and (try_num + 1) < try_count:
15                     ctx.vlog("Galaxy seems to have timedout, retrying to fetch status.")
16                     continue
17                 else:
18                     raise
19     finally:
20         gi.timeout = None
21 
22 
23 def has_jobs_in_states(ctx, gi, history_id, states):
24     params = {"history_id": history_id}
25     jobs_url = gi.url + '/jobs'
26     jobs = gi.jobs._get(url=jobs_url, params=params)
27     target_jobs = [j for j in jobs if j["state"] in states]
28     return len(target_jobs) > 0
29 
30 
31 def _wait_for_history(ctx, gi, history_id, polling_backoff=0):
32     # Used to wait for active jobs and then wait for history, but now this is called
33     # after upload is complete and after the invocation has been done scheduling - so
34     # no need to wait for active jobs anymore I think.
35 
36     def state_func():
37         return _retry_on_timeouts(ctx, gi, lambda gi: gi.histories.show_history(history_id))
38 
39     return _wait_on_state(state_func, polling_backoff)
40 
41 
42 def _wait_for_job(gi, job_id):
43     def state_func():
44         return gi.jobs.show_job(job_id, full_details=True)
45 
46     return _wait_on_state(state_func)
47 
48 
49 def _wait_on_state(state_func, polling_backoff=0):
50 
51     def get_state():
52         response = state_func()
53         state = response["state"]
54         if str(state) not in ["running", "queued", "new", "ready"]:
55             return state
56         else:
57             return None
58     timeout = 60 * 60 * 24
59     final_state = wait_on(get_state, "state", timeout, polling_backoff)
60     return final_state
61 
62 
63 __all__ = (
64     "execute",
65 )