Diff of /qiita_pet/util.py [000000] .. [879b32]

Switch to side-by-side view

--- 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