Bopytex/bopytex/scheduler.py

""" Scheduler for action to make """


from bopytex.tasks import Task


class Scheduler:
    def __init__(self, actions: dict, done: list[str] = None):
        self.actions = actions

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

        self._tasks = []

    @property
    def tasks(self):
        return self._tasks

    @property
    def doable_tasks(self):
        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):
        return {d for task in self.tasks for d in task.deps}

    @property
    def all_output(self):
        return {task.output for task in self.tasks}

    @property
    def done(self):
        return self._done

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

    def dispatch(self, task):
        """Do a task"""
        ans = self.actions[task.action](task.deps, task.args, task.output)
        return ans

    def __iter__(self):
        return self

    def __next__(self):
        return self.next()

    def next(self):
        undoable = []

        try:
            task = self._tasks.pop(0)
        except IndexError:
            raise StopIteration

        while not all([d in self.done for d in task.deps]):
            undoable.append(task)
            try:
                task = self._tasks.pop(0)
            except IndexError:
                self.append(undoable)
                raise StopIteration

        self.append(undoable)
        ans = self.dispatch(task)
        self._done.append(task.output)
        return ans

    def run(self):
        for _ in self:
            pass

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