Write pipeline attributes to json This function writes the pipeline and their attributes to a json file, that is intended to be read by resources/pipeline_graph.html to render a graphical output showing the DAG.
(self)
| 1336 | outfile_dag.close() |
| 1337 | |
| 1338 | def render_pipeline(self): |
| 1339 | """Write pipeline attributes to json |
| 1340 | |
| 1341 | This function writes the pipeline and their attributes to a json file, |
| 1342 | that is intended to be read by resources/pipeline_graph.html to render |
| 1343 | a graphical output showing the DAG. |
| 1344 | |
| 1345 | """ |
| 1346 | |
| 1347 | dict_viz = { |
| 1348 | "name": "root", |
| 1349 | "children": [] |
| 1350 | } |
| 1351 | last_of_us = {} |
| 1352 | |
| 1353 | f_tree = self._fork_tree if self._fork_tree else {1: [1]} |
| 1354 | |
| 1355 | for x, (k, v) in enumerate(f_tree.items()): |
| 1356 | for p in self.processes[1:]: |
| 1357 | |
| 1358 | if x == 0 and p.lane not in [k] + v: |
| 1359 | continue |
| 1360 | |
| 1361 | if x > 0 and p.lane not in v: |
| 1362 | continue |
| 1363 | |
| 1364 | if not p.parent_lane: |
| 1365 | lst = dict_viz["children"] |
| 1366 | else: |
| 1367 | lst = last_of_us[p.parent_lane] |
| 1368 | |
| 1369 | tooltip = { |
| 1370 | "name": "{}_{}".format(p.template, p.pid), |
| 1371 | "process": { |
| 1372 | "pid": p.pid, |
| 1373 | "input": p.input_type, |
| 1374 | "output": p.output_type if p.output_type else "None", |
| 1375 | "lane": p.lane, |
| 1376 | }, |
| 1377 | "children": [] |
| 1378 | } |
| 1379 | |
| 1380 | dir_var = "" |
| 1381 | for k2, v2 in p.directives.items(): |
| 1382 | dir_var += k2 |
| 1383 | for d in v2: |
| 1384 | try: |
| 1385 | # Remove quotes from string directives |
| 1386 | directive = v2[d].replace("'", "").replace('"', '') \ |
| 1387 | if isinstance(v2[d], str) else v2[d] |
| 1388 | dir_var += "{}: {}".format(d, directive) |
| 1389 | except KeyError: |
| 1390 | pass |
| 1391 | |
| 1392 | if dir_var: |
| 1393 | tooltip["process"]["directives"] = dir_var |
| 1394 | else: |
| 1395 | tooltip["process"]["directives"] = "N/A" |
no test coverage detected