--- a +++ b/qiita_pet/util.py @@ -0,0 +1,224 @@ +r""" +Util functions (:mod: `qiita_pet.util`) +====================================== + +..currentmodule:: qiita_pet.util + +This module provides different util functions for qiita_pet. + +Methods +------- + +..autosummary:: + :toctree: generated/ + + clean_str +""" +# ----------------------------------------------------------------------------- +# Copyright (c) 2014--, The Qiita Development Team. +# +# Distributed under the terms of the BSD 3-clause License. +# +# The full license is in the file LICENSE, distributed with this software. +# ----------------------------------------------------------------------------- +from tornado.escape import linkify as tornado_linkify, xhtml_unescape + +from qiita_core.util import execute_as_transaction +from qiita_db.reference import Reference + + +STATUS_STYLER = { + 'sandbox': + ('glyphicon glyphicon-eye-close', 'glyphicon glyphicon-lock', 'gray'), + 'awaiting_approval': + ('glyphicon glyphicon-eye-open', 'glyphicon glyphicon-lock', 'peru'), + 'private': + ('glyphicon glyphicon-eye-open', 'glyphicon glyphicon-lock', + '#3599FD'), + 'public': + ('glyphicon glyphicon-eye-open', 'glyphicon glyphicon-globe', 'green')} + +EBI_LINKIFIER = ('<a href="http://www.ebi.ac.uk/ena/data/view/{0}" ' + 'target="_blank">{0}</a>') + + +def linkify(link_template, item): + """Formats a strings into a URL using string replacement + + Paramters + --------- + link_template : str + The template for the URL. + item : list or tuple of str + The strings that will be inserted into the template + """ + return link_template.format(*item) + + +def clean_str(item): + """Converts input to string and replaces spaces with underscores + + Parameters + ---------- + item : anything convertable to string + item to convert and clean + + Returns + ------- + str + cleaned string + """ + return str(item).replace(" ", "_").replace(":", "") + + +def convert_text_html(message): + """Linkify URLs and turn newlines into <br/> for HTML""" + html = xhtml_unescape(tornado_linkify(message)) + return html.replace('\n', '<br/>') + + +@execute_as_transaction +def generate_param_str(param): + """Generate an html string with the parameter values + + Parameters + ---------- + param : BaseParameters + The parameter to generate the str + + Returns + ------- + str + The html string with the parameter set values + """ + values = param.values + ref = Reference(values['reference']) + result = ["<b>Reference:</b> %s %s" % (ref.name, ref.version)] + result.extend("<b>%s:</b> %s" % (name, value) + for name, value in values.items() + if name != 'reference') + return "<br/>".join(result) + + +def is_localhost(host): + """Verifies if the connection is local + + Parameters + ---------- + host : str + The requesting host, in general self.request.headers['host'] + + Returns + ------- + bool + True if local request + """ + localhost = ('localhost', '127.0.0.1') + return host.startswith(localhost) + + +def get_artifact_processing_status(artifact): + """Gets the processing status of the artifact + + Parameters + ---------- + artifact : qiita_db.artifact.Artifact + The artifact to get the processing status + + Returns + ------- + str, str + The processing status {'processing', 'failed', 'success', + 'Not processed'} + A summary of the jobs attached to the artifact + """ + preprocessing_status = 'Not processed' + preprocessing_status_msg = [] + for job in artifact.jobs(): + job_status = job.status + if job_status == 'error': + if preprocessing_status != 'success': + preprocessing_status = 'failed' + preprocessing_status_msg.append( + "<b>Job %s</b>: failed - %s" + % (job.id, job.log.msg)) + elif job_status == 'success': + preprocessing_status = 'success' + else: + if preprocessing_status != 'success': + preprocessing_status = 'processing' + preprocessing_status_msg.append( + "<b>Job %s</b>: %s" % (job.id, job_status)) + + if not preprocessing_status_msg: + preprocessing_status_msg = 'Not processed' + else: + preprocessing_status_msg = convert_text_html( + '<br/>'.join(preprocessing_status_msg)) + + return preprocessing_status, preprocessing_status_msg + + +def get_network_nodes_edges(graph, full_access, nodes=None, edges=None): + """Returns the JavaScript friendly representation of the graph + + Parameters + ---------- + graph : networkx.DiGraph + The artifact/jobs graph + full_access : bool + Whether the user has full access to the graph or not + nodes : list, optional + A pre-populated list of nodes. Useful for the analysis pipeline + edges : list, optional + A pre-populated list of edges. Useful for the analysis pipeline + + Returns + ------- + (list, list, int) + The list of nodes, the list of edges, and the worklfow id if there is + any job on construction + """ + nodes = nodes if nodes is not None else [] + edges = edges if edges is not None else [] + workflow_id = None + + # n[0] is the data type: job/artifact/type + # n[1] is the object + for n in graph.nodes(): + if n[0] == 'job': + # ignoring internal Jobs + if n[1].command.software.name == 'Qiita': + continue + atype = 'job' + name = n[1].command.name + status = n[1].status + wkflow = n[1].processing_job_workflow + if status == 'in_construction' and wkflow is not None: + workflow_id = wkflow.id + elif n[0] == 'artifact': + atype = n[1].artifact_type + status = 'artifact' + pp = n[1].processing_parameters + if pp is not None: + cmd = pp.command + if cmd.software.deprecated: + status = 'deprecated' + elif not cmd.active: + status = 'outdated' + if full_access or n[1].visibility == 'public': + name = '%s\n(%s)' % (n[1].name, n[1].artifact_type) + else: + continue + elif n[0] == 'type': + atype = n[1].type + name = '%s\n(%s)' % (n[1].name, n[1].type) + status = 'type' + else: + # this should never happen but let's add it just in case + raise ValueError('not valid node type: %s' % n[0]) + nodes.append((n[0], atype, n[1].id, name, status)) + + edges.extend([(n[1].id, m[1].id) for n, m in graph.edges()]) + + return nodes, edges, workflow_id