Feat: append, dispatch and __next__ (basic) scheduler

bopytex/scheduler.py

@@ -0,0 +1,29 @@
+""" Scheduler for action to make """
+
+
+class Scheduler:
+    def __init__(self, actions: list):
+        self.actions = actions
+        self._tasks = []
+        self._done = []
+
+    @property
+    def tasks(self):
+        return self._tasks
+
+    @property
+    def done(self):
+        return self._done
+
+    def append(self, tasks):
+        self._tasks += tasks
+
+    def dispatch(self, task):
+        """ Do a task """
+        ans = self.actions[task.action](task.deps)
+        return ans
+
+    def __next__(self):
+        task = self._tasks.pop()
+        self._done.append(task.output)
+        return self.dispatch(task)

test/test_scheduler.py

@@ -0,0 +1,27 @@
+from bopytex.planner import Task
+from bopytex.scheduler import Scheduler
+
+
+def test_schedule_append():
+    actions = {"DO": lambda _: "done"}
+    scheduler = Scheduler(actions)
+    tasks = [Task(action="DO", args={}, deps=[], output="")]
+    scheduler.append(tasks)
+    assert scheduler.tasks == tasks
+
+def test_schedule_dispatch():
+    actions = {"DO": lambda _: "done"}
+    scheduler = Scheduler(actions)
+    task = Task(action="DO", args={}, deps=[], output="Nothing")
+    result = scheduler.dispatch(task)
+    assert result == "done"
+
+
+def test_schedule_one_task():
+    actions = {"DO": lambda _: "done"}
+    scheduler = Scheduler(actions)
+    scheduler.append([Task(action="DO", args={}, deps=[], output="Nothing")])
+    result = scheduler.__next__()
+    assert result == "done"
+    assert scheduler.tasks == []
+    assert scheduler.done == ["Nothing"]