Merge remote-tracking branch 'origin/master' into feature/export-cwl

daliuge-common/dlg/common/__init__.py

@@ -26,7 +26,45 @@
 from .network import check_port, connect_to, portIsClosed, portIsOpen, write_to
 from .streams import ZlibCompressedStream, JSONStream
 
-STORAGE_TYPES = {'memory', 'file', 'ngas', 'null', 'json'}
+class Categories:
+    START = 'Start'
+    END = 'End'
+
+    MEMORY = 'Memory'
+    FILE = 'File'
+    NGAS = 'NGAS'
+    NULL = 'null'
+    JSON = 'json'
+    S3 = 'S3'
+
+    MKN = 'MKN'
+    SCATTER = 'Scatter'
+    GATHER = 'Gather'
+    GROUP_BY = 'GroupBy'
+    LOOP = 'Loop'
+    BRANCH = 'Branch'
+    VARIABLES = 'Variables'
+
+    DATA = 'Data'
+    COMPONENT = 'Component'
+    BASH_SHELL_APP = 'BashShellApp'
+    MPI = 'Mpi'
+    DYNLIB_APP = 'DynlibApp'
+    DOCKER = 'Docker'
+    DYNLIB_PROC_APP = 'DynlibProcApp'
+
+    COMMENT = 'Comment'
+    DESCRIPTION = 'Description'
+
+STORAGE_TYPES = {Categories.MEMORY, Categories.FILE, Categories.NGAS, Categories.NULL, Categories.JSON}
+APP_DROP_TYPES = [
+    Categories.COMPONENT,
+    Categories.BASH_SHELL_APP,
+    Categories.MPI,
+    Categories.DYNLIB_APP,
+    Categories.DOCKER,
+    Categories.DYNLIB_PROC_APP
+]
 
 if sys.version_info[0] > 2:
     def b2s(b, enc='utf8'):
@@ -143,4 +181,4 @@ def get_leaves(pg_spec):
             if dropspec.get('consumers', None) or dropspec.get('streamingConsumers', None):
                 nonleaves.add(oid)
 
-    return all_oids - nonleaves
+    return all_oids - nonleaves

daliuge-runtime/dlg/dask_emulation.py

@@ -32,7 +32,7 @@
 
 from . import utils, droputils
 from .apps import pyfunc
-from .common import dropdict
+from .common import dropdict, Categories
 from .ddap_protocol import DROPStates
 from .drop import BarrierAppDROP
 from .exceptions import InvalidDropException
@@ -333,7 +333,7 @@ def __init__(self, producer=None, pydata=_no_data):
         logger.debug("Created %r", self)
 
     def make_dropdict(self):
-        my_dropdict = dropdict({'type': 'plain', 'storage': 'memory'})
+        my_dropdict = dropdict({'type': 'plain', 'storage': Categories.MEMORY})
         if not self.producer:
             my_dropdict['pydata'] = pyfunc.serialize_data(self.pydata)
         return my_dropdict
@@ -374,4 +374,4 @@ def delayed(x, *args, **kwargs):
         nout = None
     if callable(x):
         return _AppDrop(x, nout=nout)
-    return _DataDrop(pydata=x)
+    return _DataDrop(pydata=x)