""" 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) -> list[str]: """List dependencies of all tasks""" return {d for task in self.tasks for d in task.deps} @property def all_output(self) -> list[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()