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
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245 | def task_hierarchy_to_md(
task_hierarchies_dict,
task_methods_dict,
task_inputs_dict,
task_outputs_dict,
method_params_dict,
level=0,
top_task=None,
):
"""
Convert input dictionaries into a markdown representation.
Args:
task_hierarchies_dict (dict): A dictionary representing the task hierarchy. Each key is a task IRI and each value is a dictionary (of the same structure) representing the sub-tasks of the key task.
task_methods_dict (dict): A dictionary mapping task IRIs to their associated methods.
task_inputs_dict (dict): A dictionary mapping task IRIs to their associated inputs.
task_outputs_dict (dict): A dictionary mapping task IRIs to their associated outputs.
method_params_dict (dict): A dictionary mapping method IRIs to their associated parameters.
level (int, optional): The current level of the task hierarchy. Defaults to 0.
top_task (str, optional): The top-level task in the hierarchy. Defaults to None.
Returns:
str: Markdown representation containing the hierarchy of tasks with their methods, inputs, outputs, and method parameters.
"""
if level == 0:
md_text = "🗒️ **Note**: Parent tasks are marked with 📜. Only bottom-level tasks (marked with ☑️) can be used while creating a pipeline.\n\n"
else:
md_text = ""
for task_iri, sub_tasks_dict in task_hierarchies_dict.items():
task_name = task_iri.split("#")[-1]
kg_schema_short = ""
for _, kg_schema_info in KG_SCHEMAS.items():
if task_iri.startswith(kg_schema_info["namespace"]):
kg_schema_short = kg_schema_info["namespace_prefix"]
break
kg_schema_short_text = (
f'<span style="float: right; font-weight: 100;"> 🗒️ belongs to KG schema with abbr. <code>{kg_schema_short}</code></span>'
if level == 0
else ""
)
md_text += "\t" * level + f"<details>\n"
if task_iri not in task_methods_dict: # task is a subclass of ds:Task but not of ds:AtomicTask
md_text += "\t" * (level + 1) + f"<summary>{task_name} 📜{kg_schema_short_text}</summary>\n"
md_text += "\t" * (level + 1) + f"<ul>\n"
# save top_task for getting the inputs/outputs
if level == 0:
top_task = task_iri
# handle the rest of the tasks in the hierarchy
md_text += task_hierarchy_to_md(
sub_tasks_dict,
task_methods_dict,
task_inputs_dict,
task_outputs_dict,
method_params_dict,
level + 2,
top_task,
)
else: # task has methods attached i.e. task is a subclass of ds:AtomicTask
md_text += "\t" * (level + 1) + f"<summary>{task_name} ☑️{kg_schema_short_text}</summary>\n"
md_text += "\t" * (level + 1) + f"<ul>\n"
if level == 0:
top_task = task_iri
# handle the inputs attached to the task
md_text += "\t" * (level + 2) + f"<details>\n"
md_text += "\t" * (level + 3) + f"<summary>Inputs</summary>\n"
if top_task in task_inputs_dict:
# check top task of the hierarchy
# open the list
md_text += "\t" * (level + 3) + f"<ul>\n"
for input_iri in task_inputs_dict[top_task]:
input_name = input_iri.split("#")[-1]
md_text += "\t" * (level + 4) + f"<li>{input_name}</li>\n"
if task_iri in task_inputs_dict and task_iri != top_task:
# check current task (bottom of the hierarchy)
if top_task not in task_inputs_dict:
# open the list
md_text += "\t" * (level + 3) + f"<ul>\n"
for input_iri in task_inputs_dict[task_iri]:
input_name = input_iri.split("#")[-1]
md_text += "\t" * (level + 4) + f"<li>{input_name}</li>\n"
if top_task in task_inputs_dict or (task_iri in task_inputs_dict and task_iri != top_task):
# close the list
md_text += "\t" * (level + 3) + f"</ul>\n"
md_text += "\t" * (level + 2) + f"</details>\n"
# handle the outputs attached to the task
md_text += "\t" * (level + 2) + f"<details>\n"
md_text += "\t" * (level + 3) + f"<summary>Outputs</summary>\n"
if top_task in task_outputs_dict:
# check top task of the hierarchy
# open the list
md_text += "\t" * (level + 3) + f"<ul>\n"
for output_iri in task_outputs_dict[top_task]:
output_name = output_iri.split("#")[-1]
md_text += "\t" * (level + 4) + f"<li>{output_name}</li>\n"
if task_iri in task_outputs_dict and task_iri != top_task:
# check current task (bottom of the hierarchy)
if top_task not in task_outputs_dict:
# open the list
md_text += "\t" * (level + 3) + f"<ul>\n"
for output_iri in task_outputs_dict[task_iri]:
output_name = output_iri.split("#")[-1]
md_text += "\t" * (level + 4) + f"<li>{output_name}</li>\n"
if top_task in task_outputs_dict or (task_iri in task_outputs_dict and task_iri != top_task):
# close the list
md_text += "\t" * (level + 3) + f"</ul>\n"
md_text += "\t" * (level + 2) + f"</details>\n"
# handle the methods attached to the task
md_text += "\t" * (level + 2) + f"<details>\n"
md_text += "\t" * (level + 3) + f"<summary>Methods</summary>\n"
md_text += "\t" * (level + 3) + f"<ul>\n"
for method_iri in task_methods_dict[task_iri]:
method_name = method_iri.split("#")[-1]
md_text += "\t" * (level + 3) + f"<details>\n"
md_text += "\t" * (level + 4) + f"<summary>{method_name}</summary>\n"
if method_iri in method_params_dict:
md_text += "\t" * (level + 4) + f"<ul>\n"
for param_iri, param_datatypes in method_params_dict[method_iri].items():
param_name = param_iri.split("#")[-1]
md_text += (
"\t" * (level + 5)
+ f"<li>{param_name} ({', '.join([datatype.split('#')[-1] for datatype in param_datatypes])})</li>\n"
)
md_text += "\t" * (level + 4) + f"</ul>\n"
else:
md_text += "\t" * (level + 4) + f"<ul>No parameters</ul>\n"
md_text += "\t" * (level + 3) + f"</details>\n"
md_text += "\t" * (level + 3) + f"</ul>\n"
md_text += "\t" * (level + 2) + f"</details>\n"
md_text += "\t" * (level + 1) + f"</ul>\n"
md_text += "\t" * level + f"</details>\n"
return md_text
|