Source code for nsaph.util.cwl_collect_outputs

"""
Command line utility to generate output sections from
a CWL tool or sub-workflow. 

Generated sections then can be copied and pasted into
any calling workflow

"""
#  Copyright (c) 2021. Harvard University
#
#  Developed by Research Software Engineering,
#  Faculty of Arts and Sciences, Research Computing (FAS RC)
#  Author: Michael A Bouzinier
#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.
#

import os
import sys
from argparse import ArgumentParser

import yaml


[docs]def encode(name: str): if len(name) < 13: return name + '_' else: return name[:9] + str(hash(name) % 1000) + '_'
[docs]def collect(step: str, path: str, name = None, output = sys.stdout, what=None): if what is None: what = ["step", "pipeline"] if name is None: name = step with open(path) as f: cwl = yaml.safe_load(f) outputs = cwl["outputs"] if "step" in what: print(" out:", file=output) for o in outputs: t = outputs[o]["type"] print(" - {}".format(o), file=output) if "pipeline" in what: print("## Generated by nsaph/util/cwl_collect_outputs.py from {}:". format(os.path.basename(path)), file=output) for o in outputs: t = outputs[o]["type"] if isinstance(t, str) and t.lower() in ["stdout", "stderr"]: t = "File" print(" {}:".format(encode(name) + o), file=output) print(" type: {}".format(t), file=output) print(" outputSource: {}/{}".format(step, o), file=output)
[docs]def parse_args(): parser = ArgumentParser ( description="Collect outputs from CWL sub-workflow and print" "outputs section to be added for enveloping workflow" ) parser.add_argument( dest="step", help="Step name in the outer (calling) workflow" ) parser.add_argument( dest = "path", help="Path to sub-workflow CWL file" ) parser.add_argument( "--name", help="Name to be used as output prefix, defaults to the step name", required=False ) return parser.parse_args()
if __name__ == '__main__': arguments = parse_args() collect(arguments.step, arguments.path, arguments.name)