Bopytex/bopytex/scheduler.py

""" Scheduler for action to make """


from bopytex.tasks import Task


class Scheduler:
    """Scheduler is responsible of getting tasks (the tasks) and yield those that can be done"""

    def __init__(self, done: list[str] = None):

        if done is None:
            self._done = []
        else:
            self._done = done

        self._tasks = []

    @property
    def tasks(self) -> list[Task]:
        """List all the tasks todo"""
        return self._tasks

    @property
    def doable_tasks(self) -> list[Task]:
        """List all doable tasks"""
        return [
            task
            for task in self.tasks
            if not task.deps or all([d in self.done for d in task.deps])
        ]

    @property
    def all_deps(self) -> set[str]:
        """List dependencies of all tasks"""
        return {d for task in self.tasks for d in task.deps}

    @property
    def all_output(self) -> set[str]:
        """List ouput of all tasks"""
        return {task.output for task in self.tasks}

    @property
    def done(self) -> list[str]:
        return self._done

    def append(self, tasks: list[Task]):
        self._tasks += tasks

    def is_finishable(self):
        return self.all_deps.issubset(self.all_output)

    def next_task(self):
        try:
            task = self.doable_tasks[0]
        except IndexError:
            raise StopIteration

        self._done.append(task.output)
        self._tasks.remove(task)
        return task

    def backlog(self):
        """ Yield tasks sorted according to dependencies """
        while self.doable_tasks:
            yield self.next_task()