2025 Python Packaging Survey is now live!  Take the survey now

API for graphs and tasks in Ewoks

Navigation

Verified details

These details have been verified by PyPI
Maintainers
Avatar for huderl from gravatar.com huderl Avatar for maxencer from gravatar.com maxencer Avatar for payno from gravatar.com payno Avatar for poautran from gravatar.com poautran Avatar for woutdenolf from gravatar.com woutdenolf

Unverified details

These details have not been verified by PyPI
Project links
Meta
  • License: MIT License (MIT)
  • Author: ESRF
  • Tags ewoks
  • Requires: Python >=3.6
Classifiers
Report project as malware

Project description

ewokscore

ewokscore provides an API to define workflows and implement tasks in ewoks.

Install

pip install ewokscore[test]

Test

pytest --pyargs ewokscore.tests

Getting started

from ewokscore import Task
from ewokscore import execute_graph


# Implement a workflow task
class SumTask(
    Task, input_names=["a"], optional_input_names=["b"], output_names=["result"]
):
    def run(self):
        result = self.inputs.a
        if self.inputs.b:
            result += self.inputs.b
        self.outputs.result = result


# Define a workflow with default inputs
nodes = [
    {
        "id": "task1",
        "task_type": "class",
        "task_identifier": "__main__.SumTask",
        "default_inputs": [{"name": "a", "value": 1}],
    },
    {
        "id": "task2",
        "task_type": "class",
        "task_identifier": "__main__.SumTask",
        "default_inputs": [{"name": "b", "value": 1}],
    },
    {
        "id": "task3",
        "task_type": "class",
        "task_identifier": "__main__.SumTask",
        "default_inputs": [{"name": "b", "value": 1}],
    },
]
links = [
    {
        "source": "task1",
        "target": "task2",
        "data_mapping": [{"source_output": "result", "target_input": "a"}],
    },
    {
        "source": "task2",
        "target": "task3",
        "data_mapping": [{"source_output": "result", "target_input": "a"}],
    },
]
workflow = {"graph": {"id": "testworkflow"}, "nodes": nodes, "links": links}

# Define task inputs
inputs = [{"id": "task1", "name": "a", "value": 10}]

# Execute a workflow (use a proper Ewoks task scheduler in production)
varinfo = {"root_uri": "/tmp/myresults"}  # optionally save all task outputs
result = execute_graph(workflow, varinfo=varinfo, inputs=inputs)
print(result)

Documentation

https://ewokscore.readthedocs.io/

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page