Mercurial > repos > shellac > sam_consensus_v3
view env/lib/python3.9/site-packages/gxformat2/cytoscape.py @ 0:4f3585e2f14b draft default tip
"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"
author | shellac |
---|---|
date | Mon, 22 Mar 2021 18:12:50 +0000 |
parents | |
children |
line wrap: on
line source
"""Build standalone visualization for Galaxy workflows.""" import argparse import json import os import string import sys import pkg_resources from gxformat2.model import ensure_step_position from gxformat2.normalize import steps_normalized CYTOSCAPE_JS_TEMPLATE = pkg_resources.resource_filename(__name__, 'cytoscape.html') MAIN_TS_PREFIX = "toolshed.g2.bx.psu.edu/repos/" SCRIPT_DESCRIPTION = """ This script converts the an executable Galaxy workflow (in either format - Format 2 or native .ga) into a format for visualization with Cytoscape (https://cytoscape.org/). If the target output path ends with .html this script will output a HTML page with the workflow visualized using cytoscape.js. Otherwise, this script will output a JSON description of the workflow elements for consumption by Cytoscape. """ def to_cytoscape(workflow_path: str, output_path=None): """Produce cytoscape output for supplied workflow path.""" if output_path is None: output_path, _ = os.path.splitext(workflow_path) output_path += ".html" steps = steps_normalized(workflow_path=workflow_path) elements = [] for i, step in enumerate(steps): step_id = step.get("id") or step.get("label") or str(i) step_type = step.get("type") or 'tool' classes = ["type_%s" % step_type] if step_type in ['tool', 'subworkflow']: classes.append("runnable") else: classes.append("input") tool_id = step.get("tool_id") if tool_id and tool_id.startswith(MAIN_TS_PREFIX): tool_id = tool_id[len(MAIN_TS_PREFIX):] label = step.get("id") or step.get("label") or ("tool:%s" % tool_id) or str(i) ensure_step_position(step, i) node_position = dict(x=int(step["position"]["left"]), y=int(step["position"]["top"])) repo_link = None if "tool_shed_repository" in step: repo = step["tool_shed_repository"] repo_link = "https://" + repo["tool_shed"] + "/view/" + repo["owner"] + "/" + repo["name"] + "/" + repo["changeset_revision"] node_data = { "id": step_id, "label": label, "doc": step.get("doc"), "tool_id": step.get("tool_id"), "step_type": step_type, "repo_link": repo_link } elements.append({"group": "nodes", "data": node_data, "classes": classes, "position": node_position}) for key, value in (step.get("in") or {}).items(): # handle lists? if isinstance(value, dict) and 'source' in value: value = value["source"] elif isinstance(value, dict): continue if "/" in value: from_step, output = value.split("/", 1) else: from_step, output = value, None edge_id = "%s__to__%s" % (step_id, from_step) edge_data = {"id": edge_id, "source": from_step, "target": step_id, "input": key, "output": output} elements.append({"group": "edges", "data": edge_data}) if output_path.endswith(".html"): with open(CYTOSCAPE_JS_TEMPLATE, "r") as f: template = f.read() viz = string.Template(template).safe_substitute(elements=json.dumps(elements)) with open(output_path, "w") as f: f.write(viz) else: with open(output_path, "w") as f: json.dump(elements, f) def main(argv=None): """Entry point for building Cytoscape visualizations of Galaxy workflows.""" if argv is None: argv = sys.argv[1:] args = _parser().parse_args(argv) workflow_path = args.input_path output_path = args.output_path to_cytoscape(workflow_path, output_path) def _parser(): parser = argparse.ArgumentParser(description=SCRIPT_DESCRIPTION) parser.add_argument('input_path', metavar='INPUT', type=str, help='input workflow path (.ga/gxwf.yml)') parser.add_argument('output_path', metavar='OUTPUT', type=str, nargs="?", help='output viz path (.json/.html)') return parser if __name__ == "__main__": main()