Mercurial > repos > shellac > sam_consensus_v3
comparison env/lib/python3.9/site-packages/cwltool/cwlviewer.py @ 0:4f3585e2f14b draft default tip
"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"
author | shellac |
---|---|
date | Mon, 22 Mar 2021 18:12:50 +0000 |
parents | |
children |
comparison
equal
deleted
inserted
replaced
-1:000000000000 | 0:4f3585e2f14b |
---|---|
1 """Visualize a CWL workflow.""" | |
2 import os | |
3 from urllib.parse import urlparse | |
4 | |
5 import pydot # type: ignore | |
6 import rdflib | |
7 | |
8 | |
9 class CWLViewer: | |
10 """Produce similar images with the https://github.com/common-workflow-language/cwlviewer.""" | |
11 | |
12 _queries_dir = os.path.join( | |
13 os.path.abspath(os.path.dirname(__file__)), "rdfqueries" | |
14 ) | |
15 _get_inner_edges_query_path = os.path.join(_queries_dir, "get_inner_edges.sparql") | |
16 _get_input_edges_query_path = os.path.join(_queries_dir, "get_input_edges.sparql") | |
17 _get_output_edges_query_path = os.path.join(_queries_dir, "get_output_edges.sparql") | |
18 _get_root_query_path = os.path.join(_queries_dir, "get_root.sparql") | |
19 | |
20 def __init__( | |
21 self, rdf_description # type: str | |
22 ): | |
23 """Create a viewer object based on the rdf description of the workflow.""" | |
24 self._dot_graph = CWLViewer._init_dot_graph() # type: pydot.Graph | |
25 self._rdf_graph = self._load_cwl_graph( | |
26 rdf_description | |
27 ) # type: rdflib.graph.Graph | |
28 self._root_graph_uri = self._get_root_graph_uri() # type: str | |
29 self._set_inner_edges() | |
30 self._set_input_edges() | |
31 self._set_output_edges() | |
32 | |
33 def _load_cwl_graph( | |
34 self, rdf_description # type: str | |
35 ) -> rdflib.graph.Graph: | |
36 rdf_graph = rdflib.Graph() | |
37 rdf_graph.parse(data=rdf_description, format="n3") | |
38 return rdf_graph | |
39 | |
40 def _set_inner_edges(self) -> None: | |
41 with open(self._get_inner_edges_query_path) as f: | |
42 get_inner_edges_query = f.read() | |
43 inner_edges = self._rdf_graph.query( | |
44 get_inner_edges_query, initBindings={"root_graph": self._root_graph_uri} | |
45 ) | |
46 for inner_edge_row in inner_edges: | |
47 source_label = ( | |
48 inner_edge_row["source_label"] | |
49 if inner_edge_row["source_label"] is not None | |
50 else urlparse(inner_edge_row["source_step"]).fragment | |
51 ) | |
52 n = pydot.Node( | |
53 "", | |
54 fillcolor="lightgoldenrodyellow", | |
55 style="filled", | |
56 label=source_label, | |
57 shape="record", | |
58 ) | |
59 n.set_name(str(inner_edge_row["source_step"])) | |
60 self._dot_graph.add_node(n) | |
61 target_label = ( | |
62 inner_edge_row["target_label"] | |
63 if inner_edge_row["target_label"] is not None | |
64 else urlparse(inner_edge_row["target_step"]).fragment | |
65 ) | |
66 n = pydot.Node( | |
67 "", | |
68 fillcolor="lightgoldenrodyellow", | |
69 style="filled", | |
70 label=target_label, | |
71 shape="record", | |
72 ) | |
73 n.set_name(str(inner_edge_row["target_step"])) | |
74 self._dot_graph.add_node(n) | |
75 self._dot_graph.add_edge( | |
76 pydot.Edge( | |
77 str(inner_edge_row["source_step"]), | |
78 str(inner_edge_row["target_step"]), | |
79 ) | |
80 ) | |
81 | |
82 def _set_input_edges(self) -> None: | |
83 with open(self._get_input_edges_query_path) as f: | |
84 get_input_edges_query = f.read() | |
85 inputs_subgraph = pydot.Subgraph(graph_name="cluster_inputs") | |
86 self._dot_graph.add_subgraph(inputs_subgraph) | |
87 inputs_subgraph.set_rank("same") | |
88 inputs_subgraph.create_attribute_methods(["style"]) | |
89 inputs_subgraph.set_style("dashed") | |
90 inputs_subgraph.set_label("Workflow Inputs") | |
91 | |
92 input_edges = self._rdf_graph.query( | |
93 get_input_edges_query, initBindings={"root_graph": self._root_graph_uri} | |
94 ) | |
95 for input_row in input_edges: | |
96 n = pydot.Node( | |
97 "", | |
98 fillcolor="#94DDF4", | |
99 style="filled", | |
100 label=urlparse(input_row["input"]).fragment, | |
101 shape="record", | |
102 ) | |
103 n.set_name(str(input_row["input"])) | |
104 inputs_subgraph.add_node(n) | |
105 self._dot_graph.add_edge( | |
106 pydot.Edge(str(input_row["input"]), str(input_row["step"])) | |
107 ) | |
108 | |
109 def _set_output_edges(self) -> None: | |
110 with open(self._get_output_edges_query_path) as f: | |
111 get_output_edges = f.read() | |
112 outputs_graph = pydot.Subgraph(graph_name="cluster_outputs") | |
113 self._dot_graph.add_subgraph(outputs_graph) | |
114 outputs_graph.set_rank("same") | |
115 outputs_graph.create_attribute_methods(["style"]) | |
116 outputs_graph.set_style("dashed") | |
117 outputs_graph.set_label("Workflow Outputs") | |
118 outputs_graph.set_labelloc("b") | |
119 output_edges = self._rdf_graph.query( | |
120 get_output_edges, initBindings={"root_graph": self._root_graph_uri} | |
121 ) | |
122 for output_edge_row in output_edges: | |
123 n = pydot.Node( | |
124 "", | |
125 fillcolor="#94DDF4", | |
126 style="filled", | |
127 label=urlparse(output_edge_row["output"]).fragment, | |
128 shape="record", | |
129 ) | |
130 n.set_name(str(output_edge_row["output"])) | |
131 outputs_graph.add_node(n) | |
132 self._dot_graph.add_edge( | |
133 pydot.Edge(output_edge_row["step"], output_edge_row["output"]) | |
134 ) | |
135 | |
136 def _get_root_graph_uri(self) -> rdflib.URIRef: | |
137 with open(self._get_root_query_path) as f: | |
138 get_root_query = f.read() | |
139 root = list( | |
140 self._rdf_graph.query( | |
141 get_root_query, | |
142 ) | |
143 ) | |
144 if len(root) != 1: | |
145 raise RuntimeError( | |
146 "Cannot identify root workflow! Notice that only Workflows can be visualized" | |
147 ) | |
148 | |
149 workflow = root[0]["workflow"] # type: rdflib.URIRef | |
150 return workflow | |
151 | |
152 @classmethod | |
153 def _init_dot_graph(cls) -> pydot.Graph: | |
154 graph = pydot.Graph(graph_type="digraph", simplify=False) | |
155 graph.set_bgcolor("#eeeeee") | |
156 graph.set_clusterrank("local") | |
157 graph.set_labelloc("bottom") | |
158 graph.set_labelloc("bottom") | |
159 graph.set_labeljust("right") | |
160 | |
161 return graph | |
162 | |
163 def get_dot_graph(self) -> pydot.Graph: | |
164 """Get the dot graph object.""" | |
165 return self._dot_graph | |
166 | |
167 def dot(self) -> str: | |
168 """Get the graph as graphviz.""" | |
169 return str(self._dot_graph.to_string()) |