r"""Set relationship between root task and other tasks. Two relationships are currently supported: serial and parallel. `serial` : root -> other1 -> other2 `parallel`: root -> other1 -> other2 Args: root (Task): A root task.
(
root: Task,
others: List[Task],
type: Literal["serial", "parallel"] = "parallel",
)
| 353 | |
| 354 | @staticmethod |
| 355 | def set_tasks_dependence( |
| 356 | root: Task, |
| 357 | others: List[Task], |
| 358 | type: Literal["serial", "parallel"] = "parallel", |
| 359 | ): |
| 360 | r"""Set relationship between root task and other tasks. |
| 361 | Two relationships are currently supported: serial and parallel. |
| 362 | `serial` : root -> other1 -> other2 |
| 363 | `parallel`: root -> other1 |
| 364 | -> other2 |
| 365 | |
| 366 | Args: |
| 367 | root (Task): A root task. |
| 368 | others (List[Task]): A list of tasks. |
| 369 | """ |
| 370 | # filter the root task in the others to avoid self-loop dependence. |
| 371 | others = [other for other in others if other != root] |
| 372 | |
| 373 | if len(others) == 0: |
| 374 | return |
| 375 | if type == "parallel": |
| 376 | for other in others: |
| 377 | root.add_subtask(other) |
| 378 | else: |
| 379 | parent = root |
| 380 | for child in others: |
| 381 | parent.add_subtask(child) |
| 382 | parent = child |
| 383 | |
| 384 | def add_tasks(self, tasks: Union[Task, List[Task]]) -> None: |
| 385 | r"""self.tasks and self.task_map will be updated by the input tasks.""" |
nothing calls this directly
no test coverage detected