yadage-wf.py

import time
import os
from getpass import getpass

from reana_client.api.client import (
    create_workflow_from_json,
    get_workflow_status,
    ping,
    start_workflow,
    upload_to_server,
)

if not os.getenv("REANA_SERVER_URL"):
    os.environ["REANA_SERVER_URL"] = "https://reana.cern.ch"

my_reana_token = os.getenv("REANA_ACCESS_TOKEN") or getpass("Enter your REANA token: ")

ping(my_reana_token)

my_workflow_name = "root6-roofit-yadage"

my_inputs = {
    "files": ["code/gendata.C", "code/fitdata.C"],
    "directories": ["workflow/yadage"],
    "parameters": {
        "events": "20000",
        "gendata": "code/gendata.C",
        "fitdata": "code/fitdata.C",
    },
}

my_workflow_type = "yadage"

create_workflow_from_json(
    workflow_file="workflow/yadage/workflow.yaml",
    name=my_workflow_name,
    access_token=my_reana_token,
    parameters=my_inputs,
    workflow_engine=my_workflow_type,
)


abs_path_to_input_files = [os.path.abspath(f) for f in my_inputs.get("files") or []]
upload_to_server(my_workflow_name, abs_path_to_input_files, my_reana_token)

abs_path_to_directories = [
    os.path.abspath(d) for d in my_inputs.get("directories") or []
]
upload_to_server(my_workflow_name, abs_path_to_directories, my_reana_token)

start_workflow(my_workflow_name, my_reana_token, {})

while True:
    status_details = get_workflow_status(my_workflow_name, my_reana_token)
    print("Current status: ", status_details["status"])
    if status_details["status"] == "finished":
        break
    time.sleep(10)