11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131 | class DAGExporter:
"""DAG exporter class."""
def __init__(self, config: Config) -> None:
self._config = config
def export(self, dag: nx.DiGraph, dest_dir: str, file_name: str) -> None:
"""Export DAG.
Parameters
----------
dag : nx.DiGraph
DAG.
dest_dir : str
Destination directory.
file_name : str
File name.
"""
self._export_dag(dag, dest_dir, file_name)
if self._config.figure:
self._export_fig(dag, dest_dir, file_name)
def _export_dag(self, dag: nx.DiGraph, dest_dir: str, file_name: str) -> None:
"""Export DAG description file.
Supported extension: [YAML/JSON/DOT/XML].
Parameters
----------
dag : nx.DiGraph
DAG.
dest_dir : str
Destination directory.
file_name : str
File name.
"""
if self._config.yaml:
data = json_graph.node_link_data(dag)
s = json.dumps(data)
dic = json.loads(s)
with open(f"{dest_dir}/{file_name}.yaml", "w") as f:
yaml.dump(dic, f)
if self._config.json:
data = json_graph.node_link_data(dag)
s = json.dumps(data)
with open(f"{dest_dir}/{file_name}.json", "w") as f:
json.dump(s, f)
if self._config.dot:
nx.drawing.nx_pydot.write_dot(dag, f"{dest_dir}/{file_name}.dot")
if self._config.xml:
nx.write_graphml_xml(dag, f"{dest_dir}/{file_name}.xml")
def _export_fig(self, dag: nx.DiGraph, dest_dir: str, file_name: str) -> None:
"""Export DAG figure.
Supported extension: [PNG/PDF/EPS/SVG].
Parameters
----------
dag : nx.DiGraph
DAG.
dest_dir : str
Destination directory.
file_name : str
File name.
"""
# Preprocessing
for node_i in dag.nodes():
dag.nodes[node_i]["label"] = (
f"[{node_i}]\n" f'C: {dag.nodes[node_i]["execution_time"]}'
)
if period := dag.nodes[node_i].get("period"):
dag.nodes[node_i]["shape"] = "box"
dag.nodes[node_i]["label"] += f"\nT: {period}"
if deadline := dag.nodes[node_i].get("end_to_end_deadline"):
dag.nodes[node_i]["style"] = "bold"
dag.nodes[node_i]["label"] += f"\nD: {deadline}"
for src_i, tgt_i in dag.edges():
if comm := dag.edges[src_i, tgt_i].get("communication_time"):
dag.edges[src_i, tgt_i]["label"] = f" {comm}"
dag.edges[src_i, tgt_i]["fontsize"] = 10
# Add legend
if self._config.draw_legend:
legend_str = [
"----- Legend ----\n\n",
"Circle node: Event-driven node\l",
"[i]: Task index\l",
"C: Worst-case execution time (WCET)\l",
]
if self._config.multi_rate:
legend_str.insert(1, "Square node: Timer-driven node\l")
legend_str.append("T: Period\l")
if self._config.end_to_end_deadline:
legend_str.append("D: End-to-end deadline\l")
if self._config.communication_time:
legend_str.append("Number attached to arrow: Communication time\l")
dag.add_node(-1, label="".join(legend_str), fontsize=15, shape="box3d")
# Export
pdot = nx.drawing.nx_pydot.to_pydot(dag)
if self._config.png:
pdot.write_png(f"{dest_dir}/{file_name}.png")
if self._config.svg:
pdot.write_svg(f"{dest_dir}/{file_name}.svg")
if self._config.pdf:
pdot.write_pdf(f"{dest_dir}/{file_name}.pdf")
if self._config.eps:
pdot.write_ps(f"{dest_dir}/{file_name}.ps")
subprocess.run(
f"eps2eps {dest_dir}/{file_name}.ps {dest_dir}/{file_name}.eps \
&& rm {dest_dir}/{file_name}.ps",
shell=True,
)
|