Skip to content

mecfs_bio.build_system.tasks.simple_tasks

Classes:

Functions:

  • find_tasks –

    Build a SimpleTasks object by walking the task graph

SimpleTasks

Bases: Tasks

find_tasks 

find_tasks(target_tasks: list[Task]) -> SimpleTasks

Build a SimpleTasks object by walking the task graph

Source code in mecfs_bio/build_system/tasks/simple_tasks.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def find_tasks(target_tasks: list[Task]) -> SimpleTasks:
    """
    Build a SimpleTasks object by walking the task graph
    """
    _tasks: dict[AssetId, Task] = {}
    visited = set()

    def explore_task(t: Task):
        visited.add(t.asset_id)
        for dep in t.deps:
            if dep.asset_id not in visited:
                explore_task(dep)
        if t.asset_id in _tasks and (t != _tasks[t.asset_id]):
            raise ValueError(
                f"Found two tasks with asset id  {t.asset_id}.  tasks are {t} and {_tasks[t.asset_id]}"
            )
        _tasks[t.asset_id] = t

    for task in target_tasks:
        explore_task(task)
    return SimpleTasks(_tasks)