Merge pull request #19 from ICRAR/feature/export-cwl

Feature/export cwl

.travis.yml

@@ -84,7 +84,7 @@ install:
 # run the tests, making sure subprocesses generate coverage information
 script:
  - COVFILES=
- - test -n "$NO_DLG_TRANSLATOR" || { (cd daliuge-translator && py.test --cov) && COVFILES+=" daliuge-translator/.coverage"; }
+ - test -n "$NO_DLG_TRANSLATOR" || { (cd daliuge-translator && pip install -r test-requirements.txt && py.test --cov) && COVFILES+=" daliuge-translator/.coverage"; }
  - test -n "$NO_DLG_RUNTIME" || { (cd daliuge-runtime && py.test --cov) && COVFILES+=" daliuge-runtime/.coverage"; }
  - coverage combine $COVFILES
  - test -z "$TEST_OPENAPI" || (cd OpenAPI/tests && ./test_managers_openapi.sh)

daliuge-common/dlg/clients.py

@@ -184,4 +184,4 @@ def __init__(self, host='localhost', port=constants.MASTER_DEFAULT_REST_PORT, ti
         super(MasterManagerClient, self).__init__(host=host, port=port, timeout=timeout)
 
     def create_island(self, island_host, nodes):
-        self._post_json('/managers/%s/dataisland' % (urllib.quote(island_host)), {'nodes': nodes})
+        self._post_json('/managers/%s/dataisland' % (urllib.quote(island_host)), {'nodes': nodes})

daliuge-common/dlg/common/__init__.py

@@ -69,10 +69,15 @@ class Categories:
 if sys.version_info[0] > 2:
     def b2s(b, enc='utf8'):
         return b.decode(enc)
+    def u2s(u):
+        return u
 else:
     def b2s(b, enc='utf8'):
         return b
+    def u2s(u, enc='utf-8'):
+        return u.encode(enc)
 b2s.__doc__ = "Converts bytes into a string"
+u2s.__doc__ = 'Converts text into a string'
 
 
 class dropdict(dict):

daliuge-translator/dlg/dropmake/cwl.py

@@ -0,0 +1,149 @@
+
+#    ICRAR - International Centre for Radio Astronomy Research
+#    (c) UWA - The University of Western Australia, 2015
+#    Copyright by UWA (in the framework of the ICRAR)
+#    All rights reserved
+#
+#    This library is free software; you can redistribute it and/or
+#    modify it under the terms of the GNU Lesser General Public
+#    License as published by the Free Software Foundation; either
+#    version 2.1 of the License, or (at your option) any later version.
+#
+#    This library is distributed in the hope that it will be useful,
+#    but WITHOUT ANY WARRANTY; without even the implied warranty of
+#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+#    Lesser General Public License for more details.
+#
+#    You should have received a copy of the GNU Lesser General Public
+#    License along with this library; if not, write to the Free Software
+#    Foundation, Inc., 59 Temple Place, Suite 330, Boston,
+#    MA 02111-1307  USA
+#
+
+import logging
+import os
+from zipfile import ZipFile
+
+import cwlgen
+
+from dlg import common
+
+
+#from ..common import dropdict, get_roots
+logger = logging.getLogger(__name__)
+
+
+def create_workflow(drops, pgt_path, cwl_path, zip_path):
+    """
+    Create a CWL workflow from a given Physical Graph Template
+
+    A CWL workflow consists of multiple files. A single file describing the
+    workflow, and multiple files each describing one step in the workflow. All
+    the files are combined into one zip file, so that a single file can be
+    downloaded by the user.
+
+    NOTE: CWL only supports workflow steps that are bash shell applications
+          Non-BashShellApp nodes are unable to be implemented in CWL
+    """
+
+    # create list for command line tool description files
+    step_files = []
+
+    # create the workflow
+    cwl_workflow = cwlgen.Workflow('', label='', doc='', cwl_version='v1.0')
+
+    # create files dictionary
+    files = {}
+
+    # look for input and output files in the pg_spec
+    for index, node in enumerate(drops):
+        command = node.get('command', None)
+        dataType = node.get('dt', None)
+        outputId = node.get('oid', None)
+        outputs = node.get('outputs', [])
+
+        if len(outputs) > 0:
+            files[outputs[0]] = "step" + str(index) + "/output_file_0"
+
+    # add steps to the workflow
+    for index, node in enumerate(drops):
+        dataType = node.get('dt', '')
+
+        if dataType == 'BashShellApp':
+            name = node.get('nm', '')
+            inputs = node.get('inputs', [])
+            outputs = node.get('outputs', [])
+
+            # create command line tool description
+            filename = "step" + str(index) + ".cwl"
+            filename_with_path = os.path.join(pgt_path, filename)
+            create_command_line_tool(node, filename_with_path)
+            step_files.append(filename_with_path)
+
+            # create step
+            step = cwlgen.WorkflowStep("step" + str(index), run=filename)
+
+            # add input to step
+            for index, input in enumerate(inputs):
+                step.inputs.append(cwlgen.WorkflowStepInput('input_file_' + str(index), source=files[input]))
+
+            # add output to step
+            for index, output in enumerate(outputs):
+                step.out.append(cwlgen.WorkflowStepOutput('output_file_' + str(index)))
+
+            # add step to workflow
+            cwl_workflow.steps.append(step)
+
+    # save CWL to path
+    with open(cwl_path, "w") as f:
+        f.write(cwl_workflow.export_string())
+
+    # put workflow and command line tool description files all together in a zip
+    zipObj = ZipFile(zip_path, 'w')
+    for step_file in step_files:
+        zipObj.write(step_file, os.path.basename(step_file))
+    zipObj.write(cwl_path, os.path.basename(cwl_path))
+    zipObj.close()
+
+
+def create_command_line_tool(node, filename):
+    """
+    Create a command line tool description file for a single step in a CWL
+    workflow.
+
+    NOTE: CWL only supports workflow steps that are bash shell applications
+          Non-BashShellApp nodes are unable to be implemented in CWL
+    """
+
+    # get inputs and outputs
+    inputs = node.get('inputs', [])
+    outputs = node.get('outputs', [])
+
+    # strip command down to just the basic command, with no input or output parameters
+    base_command = node.get('command', '')
+
+    # TODO: find a better way of specifying command line program + arguments
+    base_command = base_command[:base_command.index(" ")]
+    base_command = common.u2s(base_command)
+
+    # cwlgen's Serializer class doesn't support python 2.7's unicode types
+    cwl_tool = cwlgen.CommandLineTool(tool_id=common.u2s(node['app']), label=common.u2s(node['nm']), base_command=base_command, cwl_version='v1.0')
+
+    # add inputs
+    for index, input in enumerate(inputs):
+        file_binding = cwlgen.CommandLineBinding(position=index)
+        input_file = cwlgen.CommandInputParameter('input_file_' + str(index), param_type='File', input_binding=file_binding, doc='input file ' + str(index))
+        cwl_tool.inputs.append(input_file)
+
+    if len(inputs) == 0:
+        cwl_tool.inputs.append(cwlgen.CommandInputParameter('dummy', param_type='null', doc='dummy'))
+
+    # add outputs
+    for index, output in enumerate(outputs):
+        file_binding = cwlgen.CommandLineBinding()
+        output_file = cwlgen.CommandOutputParameter('output_file_' + str(index), param_type='stdout', output_binding=file_binding, doc='output file ' + str(index))
+        cwl_tool.outputs.append(output_file)
+
+    # write to file
+    with open(filename, "w") as f:
+        f.write(cwl_tool.export_string())

daliuge-translator/dlg/dropmake/dm_utils.py

@@ -57,7 +57,7 @@ def get_lg_ver_type(lgo):
             for fd in fds:
                 if "name" in fd:
                     kw = fd["name"]
-                    if kw in node:
+                    if kw in node and kw not in ['description']:
                         return LG_VER_EAGLE_CONVERTED
                     else:
                         return LG_VER_EAGLE

daliuge-translator/dlg/dropmake/web/lg_web.py

@@ -51,6 +51,7 @@
 from ..pg_generator import unroll, partition, GraphException
 from ..pg_manager import PGManager
 from ..scheduler import SchedulerException
+from ..cwl import create_workflow
 
 
 def file_as_string(fname, enc="utf8"):
@@ -197,6 +198,37 @@ def pgtjsonbody_get():
         response.status = 404
         return "{0}: JSON graph {1} not found\n".format(err_prefix, pgt_name)
 
+@get("/pgt_cwl")
+def pgtcwl_get():
+    """
+    Return CWL representation of the logical graph
+    """
+    pgt_name = request.query.get("pgt_name")
+
+    if pgt_exists(pgt_name):
+        # get PGT from manager
+        pgtp = pg_mgr.get_pgt(pgt_name)
+
+        # debug
+        print("pgtp:" + str(pgtp) + ":" + str(dir(pgtp)))
+
+        # build filename for CWL file from PGT filename
+        cwl_filename = pgt_name[:-6] + ".cwl"
+        zip_filename = pgt_name[:-6] + ".zip"
+
+        # get paths used while creating the CWL files
+        root_path = pgt_path("")
+        cwl_path = pgt_path(cwl_filename)
+        zip_path = pgt_path(zip_filename)
+
+        # create the CWL workflow
+        create_workflow(pgtp.drops, root_path, cwl_path, zip_path);
+
+        # respond with download of ZIP file
+        return static_file(zip_filename, root=root_path, download=True)
+    else:
+        response.status = 404
+        return "{0}: JSON graph {1} not found\n".format(err_prefix, pgt_name)
 
 @get("/lg_editor")
 def load_lg_editor():

daliuge-translator/dlg/dropmake/web/pg_viewer.html

@@ -481,6 +481,24 @@
     }
   }
 
+  function makeCWL() {
+    //given a logical graph name, get its CWL from the server
+    $.ajax({
+        url: "/pgt_cwl?pgt_name={{pgt_view_json_name}}",
+        type: 'get',
+        error: function(XMLHttpRequest, textStatus, errorThrown) {
+            if (404 == XMLHttpRequest.status) {
+              console.error('Server cannot locate physical graph file {{pgt_view_json_name}}');
+            } else {
+              console.error('status:' + XMLHttpRequest.status + ', status text: ' + XMLHttpRequest.statusText);
+            }
+        },
+        success: function(data){
+          console.log("success", data);
+        }
+    });
+  }
+
   function zoomToFit() {
     myDiagram.zoomToFit()
     // console.log(myDiagram.viewportBounds.width.toString());
@@ -552,6 +570,7 @@
     <button id="gantt_button" class="button" onclick="genGanttChart()">Produce Gantt Chart</button>
     <button id="schedule_button" class="button" onclick="genScheduleChart()">Produce Schedule Matrix</button>
     <button id="png_button" class="button" onclick="makePNG()">Export to PNG</button>
+    <button id="cwl_button" class="button" onclick="window.location.href='/pgt_cwl?pgt_name={{pgt_view_json_name}}'">Export to CWL</button>
     <button id="zoom_button" class="button" onclick="zoomToFit()">Zoom to Fit</button>
   </div>
 </div>

daliuge-translator/dlg/translator/tool_commands.py

@@ -284,11 +284,29 @@ def dlg_submit(parser, args):
     with _open_i(opts.pg_path) as f:
         submit(json.load(f), opts)
 
+
+def cwl(parser, args):
+    tool.add_logging_options(parser)
+    _add_output_options(parser)
+    parser.add_option('-P', '--physical-graph-template', action='store', dest='pgt_path', type='string',
+                      help='Path to the Physical Graph Template (default: stdin)', default='-')
+    (opts, args) = parser.parse_args(args)
+    tool.setup_logging(opts)
+
+    # load the pgt
+    with _open_i(opts.pgt_path) as fi:
+        pgt = json.load(fi)
+
+    # create the CWL workflow
+    from ..dropmake.cwl import create_workflow
+    create_workflow(pgt, "", "workflow.cwl", opts.output)
+
 def register_commands():
     tool.cmdwrap('lgweb', 'A Web server for the Logical Graph Editor', 'dlg.dropmake.web.lg_web:run')
     tool.cmdwrap('submit', 'Submits a Physical Graph to a Drop Manager', dlg_submit)
     tool.cmdwrap('map', 'Maps a Physical Graph Template to resources and produces a Physical Graph', dlg_map)
     tool.cmdwrap('unroll', 'Unrolls a Logical Graph into a Physical Graph Template', dlg_unroll)
     tool.cmdwrap('partition', 'Divides a Physical Graph Template into N logical partitions', dlg_partition)
     tool.cmdwrap('unroll-and-partition', 'unroll + partition', dlg_unroll_and_partition)
-    tool.cmdwrap('fill', 'Fill a Logical Graph with parameters', fill)
+    tool.cmdwrap('fill', 'Fill a Logical Graph with parameters', fill)
+    tool.cmdwrap('cwl', 'Translate a Logical Graph into a Common Workflow Language (CWL) workflow', cwl)

daliuge-translator/setup.py

@@ -83,6 +83,7 @@ def write_version_info():
 
 install_requires = [
     "bottle",
+    "cwlgen",
     "daliuge-common==%s" % (VERSION,),
     "metis>=0.2a3",
     # Python 3.6 is only supported in NetworkX 2 and above

daliuge-translator/test-requirements.txt

@@ -0,0 +1,11 @@
+cwltool
+gitpython
+# cwltool's dependency breaks cwlgen's in python 2.7,
+# so we explicitly need to install a version that works
+# for both
+# A similar situation happens with typing, with pip failing
+# to automatically upgrade to >=3.7.4 when an installed
+# "typing" module satisfies a previous, more relaxed
+# constraint (e.g., >=3.5)
+ruamel.yaml==0.16.0; python_version=='2.7'
+typing>=3.7.4

daliuge-translator/test/dropmake/test_pg_gen.py

@@ -23,6 +23,9 @@
 
 from dlg.dropmake.pg_generator import LG, PGT, MetisPGTP, MySarkarPGTP,\
  MinNumPartsPGTP, GPGTNoNeedMergeException
+from dlg.dropmake.cwl import create_workflow
+from dlg.dropmake import pg_generator
+from dlg.translator.tool_commands import unroll
 
 """
 python -m unittest test.dropmake.test_pg_gen
@@ -162,3 +165,60 @@ def test_pg_eagle(self):
             fp = get_lg_fname(lg)
             lg = LG(fp)
             lg.unroll_to_tpl()
+
+    def test_cwl_translate(self):
+        import git
+        import os
+        import shutil
+        import uuid
+        import zipfile
+        import subprocess
+        import tempfile
+
+        output_list = []
+
+        # create a temporary directory to contain files created during test
+        cwl_output = tempfile.mkdtemp()
+
+        # create a temporary directory to contain a clone of EAGLE_test_repo
+        direct = tempfile.mkdtemp()
+
+        REPO = "https://github.com/ICRAR/EAGLE_test_repo"
+        git.Git(direct).clone(REPO)
+
+        cwl_dir = os.getenv("CWL_GRAPHS", "SP-602")
+
+        graph_dir = direct + "/EAGLE_test_repo/" + cwl_dir + "/"
+        for subdir, dirs, files in os.walk(graph_dir):
+            for file in files:
+                f = os.path.join(subdir, file)
+                if not f.endswith(".graph"):
+                    continue
+                pgt = unroll(f, 1)
+                pg_generator.partition(pgt, algo='metis', num_partitions=1,
+                                            num_islands=1, partition_label='partition')
+
+                uid = str(uuid.uuid4())
+                cwl_output_dir = cwl_output + '/' + uid
+                os.mkdir(cwl_output_dir)
+                cwl_out = cwl_output_dir + '/workflow.cwl'
+                cwl_out_zip = cwl_output_dir + '/workflow.zip'
+                output_list.append((cwl_out, cwl_out_zip))
+
+                create_workflow(pgt, "", cwl_out, cwl_out_zip)
+
+        for out, zip in output_list:
+            zip_ref = zipfile.ZipFile(zip)
+            zip_ref.extractall(os.path.dirname(zip))
+            zip_ref.close()
+
+            cmd = ['cwltool', '--validate', out]
+            p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+            stdout, stderr = p.communicate()
+            self.assertEqual(p.returncode, 0, b'stdout:\n' + stdout + b'\nstderr:\n' + stderr)
+
+        # delete the clone of EAGLE_test_repo
+        shutil.rmtree(direct, ignore_errors=True)
+
+        # delete the temporary output directory
+        shutil.rmtree(cwl_output, ignore_errors=True)