Mercurial > repos > shellac > sam_consensus_v3

comparison env/lib/python3.9/site-packages/gxformat2/cytoscape.py @ 0:4f3585e2f14b draft default tip

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"
author shellac
date Mon, 22 Mar 2021 18:12:50 +0000 (2021-03-22)
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:4f3585e2f14b
1 """Build standalone visualization for Galaxy workflows."""
2 import argparse
3 import json
4 import os
5 import string
6 import sys
7
8 import pkg_resources
9
10 from gxformat2.model import ensure_step_position
11 from gxformat2.normalize import steps_normalized
12
13 CYTOSCAPE_JS_TEMPLATE = pkg_resources.resource_filename(__name__, 'cytoscape.html')
14 MAIN_TS_PREFIX = "toolshed.g2.bx.psu.edu/repos/"
15 SCRIPT_DESCRIPTION = """
16 This script converts the an executable Galaxy workflow (in either format -
17 Format 2 or native .ga) into a format for visualization with Cytoscape
18 (https://cytoscape.org/).
19
20 If the target output path ends with .html this script will output a HTML
21 page with the workflow visualized using cytoscape.js. Otherwise, this script
22 will output a JSON description of the workflow elements for consumption by
23 Cytoscape.
24 """
25
26
27 def to_cytoscape(workflow_path: str, output_path=None):
28 """Produce cytoscape output for supplied workflow path."""
29 if output_path is None:
30 output_path, _ = os.path.splitext(workflow_path)
31 output_path += ".html"
32
33 steps = steps_normalized(workflow_path=workflow_path)
34 elements = []
35 for i, step in enumerate(steps):
36 step_id = step.get("id") or step.get("label") or str(i)
37 step_type = step.get("type") or 'tool'
38 classes = ["type_%s" % step_type]
39 if step_type in ['tool', 'subworkflow']:
40 classes.append("runnable")
41 else:
42 classes.append("input")
43
44 tool_id = step.get("tool_id")
45 if tool_id and tool_id.startswith(MAIN_TS_PREFIX):
46 tool_id = tool_id[len(MAIN_TS_PREFIX):]
47 label = step.get("id") or step.get("label") or ("tool:%s" % tool_id) or str(i)
48 ensure_step_position(step, i)
49 node_position = dict(x=int(step["position"]["left"]), y=int(step["position"]["top"]))
50 repo_link = None
51 if "tool_shed_repository" in step:
52 repo = step["tool_shed_repository"]
53 repo_link = "https://" + repo["tool_shed"] + "/view/" + repo["owner"] + "/" + repo["name"] + "/" + repo["changeset_revision"]
54 node_data = {
55 "id": step_id,
56 "label": label,
57 "doc": step.get("doc"),
58 "tool_id": step.get("tool_id"),
59 "step_type": step_type,
60 "repo_link": repo_link
61 }
62 elements.append({"group": "nodes", "data": node_data, "classes": classes, "position": node_position})
63 for key, value in (step.get("in") or {}).items():
64 # handle lists?
65 if isinstance(value, dict) and 'source' in value:
66 value = value["source"]
67 elif isinstance(value, dict):
68 continue
69 if "/" in value:
70 from_step, output = value.split("/", 1)
71 else:
72 from_step, output = value, None
73 edge_id = "%s__to__%s" % (step_id, from_step)
74 edge_data = {"id": edge_id, "source": from_step, "target": step_id, "input": key, "output": output}
75 elements.append({"group": "edges", "data": edge_data})
76
77 if output_path.endswith(".html"):
78 with open(CYTOSCAPE_JS_TEMPLATE, "r") as f:
79 template = f.read()
80 viz = string.Template(template).safe_substitute(elements=json.dumps(elements))
81 with open(output_path, "w") as f:
82 f.write(viz)
83 else:
84 with open(output_path, "w") as f:
85 json.dump(elements, f)
86
87
88 def main(argv=None):
89 """Entry point for building Cytoscape visualizations of Galaxy workflows."""
90 if argv is None:
91 argv = sys.argv[1:]
92
93 args = _parser().parse_args(argv)
94
95 workflow_path = args.input_path
96 output_path = args.output_path
97 to_cytoscape(workflow_path, output_path)
98
99
100 def _parser():
101 parser = argparse.ArgumentParser(description=SCRIPT_DESCRIPTION)
102 parser.add_argument('input_path', metavar='INPUT', type=str,
103 help='input workflow path (.ga/gxwf.yml)')
104 parser.add_argument('output_path', metavar='OUTPUT', type=str, nargs="?",
105 help='output viz path (.json/.html)')
106 return parser
107
108
109 if __name__ == "__main__":
110 main()