Visualizes the task graph of the passed job in the DOT format. The result is written to a file located at `path`. Note: this function requires the `pydot` package to be installed.
(job: Job, path: GenericPath)
| 5 | |
| 6 | |
| 7 | def visualize_job(job: Job, path: GenericPath): |
| 8 | """ |
| 9 | Visualizes the task graph of the passed job in the DOT format. |
| 10 | The result is written to a file located at `path`. |
| 11 | |
| 12 | Note: this function requires the `pydot` package to be installed. |
| 13 | """ |
| 14 | |
| 15 | try: |
| 16 | import pydot |
| 17 | except ImportError: |
| 18 | raise MissingPackageException("pydot") |
| 19 | |
| 20 | graph = pydot.Dot("job", graph_type="digraph") |
| 21 | visited = {} |
| 22 | |
| 23 | def visit(task: Task): |
| 24 | nonlocal visited, graph |
| 25 | if task.task_id in visited: |
| 26 | return visited[task.task_id] |
| 27 | |
| 28 | node = pydot.Node(task.label) |
| 29 | graph.add_node(node) |
| 30 | for dep in task.dependencies: |
| 31 | dep_node = visit(dep) |
| 32 | edge = pydot.Edge(dep_node.get_name(), node.get_name()) |
| 33 | graph.add_edge(edge) |
| 34 | |
| 35 | visited[task.task_id] = node |
| 36 | return node |
| 37 | |
| 38 | for task in job.tasks: |
| 39 | visit(task) |
| 40 | |
| 41 | graph.write(path) |