Yan 968

.travis.yml

@@ -34,6 +34,7 @@ matrix:
       env: NO_DLG_RUNTIME=1
     - python: "3.8"
       env: NO_DLG_TRANSLATOR=1
+    - python: "3.9"
     # NOTE: The OpenAPI code still needs to be removed
     # - python: "3.8"
     #   env: TEST_OPENAPI=1

README.rst

@@ -1,8 +1,8 @@
 Data Activated 流 Graph Engine
 ==============================
 
-.. image:: https://travis-ci.org/ICRAR/daliuge.svg?branch=master
-    :target: https://travis-ci.org/ICRAR/daliuge
+.. image:: https://travis-ci.com/ICRAR/daliuge.svg?branch=master
+    :target: https://travis-ci.com/github/ICRAR/daliuge
 
 .. image:: https://coveralls.io/repos/github/ICRAR/daliuge/badge.svg?branch=master
     :target: https://coveralls.io/github/ICRAR/daliuge?branch=master

daliuge-common/build_common.sh

@@ -19,7 +19,7 @@ case "$1" in
     "devcuda")
         export VCS_TAG=`git rev-parse --abbrev-ref HEAD | tr '[:upper:]' '[:lower:]'`
         echo "Building daliuge-common development version using tag ${VCS_TAG}"
-        docker build --build-arg VCS_TAG=${VCS_TAG} --no-cache -t icrar/daliuge-common:${VCS_TAG} -f docker/Dockerfile.cuda .
+        docker build --build-arg VCS_TAG=${VCS_TAG} --no-cache -t icrar/daliuge-common:${VCS_TAG} -f docker/Dockerfile.devcuda .
         echo "Build finished!"
         exit 0;;
     "casa")

daliuge-common/dlg/__init__.py

@@ -22,3 +22,10 @@
 
 # Declaring this as a namespace package
 __path__ = __import__("pkgutil").extend_path(__path__, __name__)  # @ReservedAssignment
+# set the version
+try:
+    from dlg.common import version
+    __version__ = version.full_version
+except:
+    # This can happen when running from source
+    __version__ = 'unknown'

daliuge-common/docker/Dockerfile.cuda → daliuge-common/docker/Dockerfile.devcuda

@@ -7,23 +7,18 @@ FROM ubuntu:20.04
 ARG BUILD_ID
 LABEL stage=builder
 LABEL build=$BUILD_ID
-RUN apt-get update && apt-get install -y gcc python3 python3.8-venv && apt-get clean
+RUN apt-get update && \
+    apt-get install -y gcc python3 python3.8-venv python3-pip python3-distutils libmetis-dev curl git sudo && \
+    apt-get clean
 
 COPY / /daliuge
 
-RUN cd && python3 -m venv dlg && cd /daliuge && \
-    . ${HOME}/dlg/bin/activate && \
-    pip install numpy && \
-    pip install . && \
-    apt-get remove -y gcc && \
-    apt-get autoremove -y
+RUN cd / && python3 -m venv dlg && cd /daliuge && \
+    . /dlg/bin/activate && \
+    pip install wheel numpy && \
+    pip install .
 
-
-FROM ubuntu:20.04
-RUN apt-get update && apt-get install -y bash
-COPY --from=0 /root/dlg /root/dlg
-
-RUN apt install -y wget gnupg2 software-properties-common
+RUN DEBIAN_FRONTEND=noninteractive apt install -y wget gnupg2 software-properties-common
 RUN mkdir -p /code && cd /code &&\
     wget https://developer.download.nvidia.com/compute/cuda/repos/ubuntu2004/x86_64/cuda-ubuntu2004.pin &&\
     mv cuda-ubuntu2004.pin /etc/apt/preferences.d/cuda-repository-pin-600 &&\

daliuge-engine/build_engine.sh

@@ -13,14 +13,15 @@ case "$1" in
         echo "Build finished!"
         exit 0 ;;
     "dev")
+        C_TAG="master"
         [[ ! -z $2 ]] && C_TAG=$2
         export VERSION=`git describe --tags --abbrev=0|sed s/v//`
         export VCS_TAG=`git rev-parse --abbrev-ref HEAD | tr '[:upper:]' '[:lower:]'`
-        echo "Building daliuge-engine development version using daliuge-common:${VCS_TAG}"
+        echo "Building daliuge-engine development version using daliuge-common:${C_TAG}"
         echo "$VERSION:$VCS_TAG" > dlg/manager/web/VERSION
         git rev-parse --verify HEAD >> dlg/manager/web/VERSION
         cp ../LICENSE dlg/manager/web/.
-        docker build --build-arg VCS_TAG=${VCS_TAG} --no-cache -t icrar/daliuge-engine:${VCS_TAG} -f docker/Dockerfile.dev .
+        docker build --build-arg VCS_TAG=${C_TAG} --no-cache -t icrar/daliuge-engine:${VCS_TAG} -f docker/Dockerfile.dev .
         echo "Build finished!"
         exit 0;;
     "devall")

daliuge-engine/dlg/__init__.py

@@ -22,3 +22,10 @@
 
 # Declaring this as a namespace package
 __path__ = __import__("pkgutil").extend_path(__path__, __name__)  # @ReservedAssignment
+# set the version
+try:
+    from dlg.common import version
+    __version__ = version.full_version
+except:
+    # This can happen when running from source
+    __version__ = 'unknown'

daliuge-engine/dlg/apps/dockerapp.py

@@ -74,7 +74,7 @@ def waitForIp(self, timeout=None):
 
 ##
 # @brief Docker
-# @details
+# @details A component wrapping docker based applications.
 # @par EAGLE_START
 # @param category Docker
 # @param tag template
@@ -93,7 +93,7 @@ def waitForIp(self, timeout=None):
 # @param[in] cparam/command_line_arguments Command Line Arguments//String/readwrite/False//False/
 #     \~English Additional command line arguments to be added to the command line to be executed
 # @param[in] cparam/paramValueSeparator Param value separator/ /String/readwrite/False//False/
-#     \~English Separator character(s) between parameters on the command line
+#     \~English Separator character(s) between parameters and their respective values on the command line
 # @param[in] cparam/argumentPrefix Argument prefix/"--"/String/readwrite/False//False/
 #     \~English Prefix to each keyed argument on the command line
 # @param[in] cparam/execution_time Execution Time/5/Float/readonly/False//False/

daliuge-engine/dlg/apps/pyfunc.py

@@ -65,13 +65,17 @@ def serialize_func(f):
 
     fser = dill.dumps(f)
     fdefaults = {"args":[], "kwargs": {}}
+    adefaults = {"args":[], "kwargs": {}}
     a = inspect.getfullargspec(f)
     if a.defaults:
         fdefaults["kwargs"] = dict(
             zip(a.args[-len(a.defaults):], [serialize_data(d) for d in a.defaults])
         )
+        adefaults["kwargs"] = dict(
+            zip(a.args[-len(a.defaults):], [d for d in a.defaults])
+        )
     logger.debug(f"Introspection of function {f}: {a}")
-    logger.debug("Defaults for function %r: %r", f, fdefaults)
+    logger.debug("Defaults for function %r: %r", f, adefaults)
     return fser, fdefaults
 
 
@@ -118,14 +122,14 @@ def import_using_code(code):
 # @param[in] cparam/group_start Group start/False/Boolean/readwrite/False//False/
 #     \~English Is this node the start of a group?
 # @param[in] cparam/input_error_threshold "Input error rate (%)"/0/Integer/readwrite/False//False/
-#     \~English the allowed failure rate of the inputs (in percent), before this component goes to ERROR state and is not executed
+#     \~English The allowed failure rate of the inputs (in percent), before this component goes to ERROR state and is not executed
 # @param[in] cparam/n_tries Number of tries/1/Integer/readwrite/False//False/
 #     \~English Specifies the number of times the 'run' method will be executed before finally giving up
 # @param[in] aparam/func_name Function Name//String/readwrite/False//False/
-#     \~English Python fuction name
+#     \~English Python function name
 # @param[in] aparam/func_code Function Code//String/readwrite/False//False/
-#     \~English Python fuction code, e.g. 'def fuction_name(args): return args'
-# @param[in] aparam/pickle Pickle//Boolean/readwrite/False//False/
+#     \~English Python function code, e.g. 'def function_name(args): return args'
+# @param[in] aparam/pickle Pickle/false/Boolean/readwrite/False//False/
 #     \~English Whether the python arguments are pickled.
 # @param[in] aparam/func_defaults Function Defaults//String/readwrite/False//False/
 #     \~English Mapping from argname to default value. Should match only the last part of the argnames list.
@@ -363,7 +367,7 @@ def run(self):
         #     #raise ValueError
 
         # use explicit mapping of inputs to arguments first
-        # TODO: Required by dlg_delayed?? Else, we should reall not do this. 
+        # TODO: Required by dlg_delayed?? Else, we should really not do this. 
         kwargs = {
             name: inputs.pop(uid)
             for name, uid in self.func_arg_mapping.items()
@@ -382,7 +386,7 @@ def run(self):
         kwargs = {}
         if ('inputs' in self.parameters and isinstance(self.parameters['inputs'][0], dict)):
             logger.debug(f"Using named ports to identify inputs: "+\
-                    f"{self.parameters['inputs']}")            
+                    f"{self.parameters['inputs']}")
             for i in range(min(len(inputs),self.fn_nargs +\
                 len(self.arguments.kwonlyargs))):
                 # key for final dict is value in named ports dict
@@ -397,25 +401,75 @@ def run(self):
         logger.debug(f"updating funcargs with {kwargs}")
         self.funcargs.update(kwargs)
 
+        # Try to get values for still missing positional arguments from Application Args
+        if "applicationArgs" in self.parameters:
+            kwargs = {}
+            posargs = self.arguments.args[:self.fn_npos]
+            for pa in posargs:
+                if pa not in self.funcargs:
+                    if pa in self.parameters["applicationArgs"]:
+                        value = self.parameters["applicationArgs"][pa]['value']
+                        ptype = self.parameters["applicationArgs"][pa]['type']
+                        if ptype in ["Complex", "Json"]:
+                            try:
+                                value = ast.literal_eval(value)
+                            except:
+                                pass
+                        kwargs.update({
+                            pa:
+                            value
+                        })
+                    else:
+                        logger.warning(f"Required positional argument '{pa}' not found!")
+            logger.debug(f"updating funcargs with {kwargs}")
+            self.funcargs.update(kwargs)
+
+            # Try to get values for still missing kwargs arguments from parameters
+            kwargs = {}
+            kws = self.arguments.args[self.fn_npos:]
+            for ka in kws:
+                if ka not in self.funcargs:
+                    if ka in self.parameters["applicationArgs"]:
+                        value = self.parameters["applicationArgs"][ka]['value']
+                        ptype = self.parameters["applicationArgs"][ka]['type']
+                        if ptype in ["Complex", "Json"]:
+                            try:
+                                value = ast.literal_eval(value)
+                            except:
+                                pass
+                        kwargs.update({
+                            ka:
+                            value
+                        })
+                    else:
+                        logger.warning(f"Keyword argument '{ka}' not found!")
+            logger.debug(f"updating funcargs with {kwargs}")
+            self.funcargs.update(kwargs)
+
         # Fill rest with default arguments if there are any more
         kwargs = {}
         for kw in self.func_defaults.keys():
+            value = self.func_defaults[kw]
             if kw not in self.funcargs:
-                kwargs.update({kw: self.func_defaults[kw]})
+                kwargs.update({kw: value})
         logger.debug(f"updating funcargs with {kwargs}")
         self.funcargs.update(kwargs)
 
         logger.debug(f"Running {self.func_name} with {self.funcargs}")
         result = self.f(**self.funcargs)
+        logger.debug(f"Finished execution of {self.func_name}.")
 
         # Depending on how many outputs we have we treat our result
         # as an iterable or as a single object. Each result is pickled
         # and written to its corresponding output
         outputs = self.outputs
-        if len(outputs) == 1:
-            result = [result]
-        for r, o in zip(result, outputs):
-            if self.pickle:
-                o.write(pickle.dumps(r))  # @UndefinedVariable
-            else:
-                o.write(repr(r).encode('utf-8'))
+        if len(outputs) > 0:
+            if len(outputs) == 1:
+                result = [result]
+            for r, o in zip(result, outputs):
+                p = pickle.dumps(r)
+                if self.pickle:
+                    logger.debug(f"Writing pickeled result {type(r)} to {o}")
+                    o.write(pickle.dumps(r))  # @UndefinedVariable
+                else:
+                    o.write(repr(r).encode('utf-8'))