soorena62
45 lines
1.5 KiB
Python
45 lines
1.5 KiB
Python
from typing import List
|
|
from radiuma_api.task import Task, Status, TaskEvent
|
|
from radiuma_api.execution_context import ExecutionContext
|
|
from radiuma_api.scheduler import Scheduler
|
|
from radiuma_api.io_port import CompatibilityException
|
|
# Write Codes Here:
|
|
|
|
|
|
class Workflow(Task):
|
|
"""Composite node: aggregates Tasks (Modules or other Workflows)."""
|
|
|
|
def __init__(self, name: str):
|
|
super().__init__(name)
|
|
self.children: List[Task] = []
|
|
|
|
def add_child(self, task: Task):
|
|
task.parent_task = self
|
|
self.children.append(task)
|
|
|
|
def run(self, context: ExecutionContext):
|
|
"""Delegate execution of children to Scheduler."""
|
|
self._set_status(Status.RUNNING)
|
|
scheduler = Scheduler()
|
|
|
|
for child in self.children:
|
|
if self.status == Status.STOPPED:
|
|
break
|
|
|
|
try:
|
|
if child.check_inputs_ready(context):
|
|
scheduler.run(child, context)
|
|
else:
|
|
print(f"Task {child.name} not ready, skipping.")
|
|
except CompatibilityException as e:
|
|
self._emit(TaskEvent.ON_ERROR, {"error": e.reason, "task": child.name})
|
|
child._set_status(Status.FAILED)
|
|
|
|
# Final status evaluation
|
|
if all(c.status == Status.COMPLETED for c in self.children):
|
|
self._set_status(Status.COMPLETED)
|
|
elif any(c.status == Status.FAILED for c in self.children):
|
|
self._set_status(Status.FAILED)
|
|
else:
|
|
self._set_status(Status.STOPPED)
|