Merge pull request #195 from ICRAR/liu-282

Liu 282

daliuge-engine/dlg/apps/pyfunc.py

@@ -29,6 +29,7 @@
 import inspect
 import json
 import logging
+import os
 import pickle
 
 from typing import Callable
@@ -272,6 +273,11 @@ def initialize(self, **kwargs):
         """
         BarrierAppDROP.initialize(self, **kwargs)
 
+        env = os.environ.copy()
+        env.update({"DLG_UID": self._uid})
+        if self._dlg_session:
+            env.update({"DLG_SESSION_ID": self._dlg_session.sessionId})
+
         self._applicationArgs = self._popArg(kwargs, "applicationArgs", {})
 
         self.func_code = self._popArg(kwargs, "func_code", None)

daliuge-engine/dlg/data/directorycontainer.py

@@ -19,6 +19,7 @@
 #    Foundation, Inc., 59 Temple Place, Suite 330, Boston,
 #    MA 02111-1307  USA
 #
+import logging
 import os
 import shutil
 
@@ -28,7 +29,24 @@
 from dlg.exceptions import InvalidDropException, InvalidRelationshipException
 from dlg.meta import dlg_bool_param
 
+logger = logging.getLogger(__name__)
 
+##
+# TODO: This needs some more work
+# @brief Directory
+# @details A ContainerDROP that represents a filesystem directory. It only allows
+# FileDROPs and DirectoryContainers to be added as children. Children
+# can only be added if they are placed directly within the directory
+# represented by this DirectoryContainer.
+# @par EAGLE_START
+# @param category Directory
+# @param tag future
+# @param data_volume Data volume/5/Float/ComponentParameter/readwrite//False/False/Estimated size of the data contained in this node
+# @param group_end Group end/False/Boolean/ComponentParameter/readwrite//False/False/Is this node the end of a group?
+# @param check_exists Check path exists/True/Boolean/ComponentParameter/readwrite//False/False/Perform a check to make sure the file path exists before proceeding with the application
+# @param dirname Directory name//String/ComponentParameter/readwrite//False/False/"Directory name/path"
+# @param dummy dummy//String/OutputPort/readwrite//False/False/Dummy output port
+# @par EAGLE_END
 class DirectoryContainer(PathBasedDrop, ContainerDROP):
     """
     A ContainerDROP that represents a filesystem directory. It only allows
@@ -49,7 +67,8 @@ def initialize(self, **kwargs):
 
         directory = kwargs["dirname"]
 
-        if self.check_exists is True:
+        logger.debug("Checking existence of %s %s", directory, self.check_exists)
+        if "check_exists" in kwargs and kwargs["check_exists"] is True:
             if not os.path.isdir(directory):
                 raise InvalidDropException(self, "%s is not a directory" % (directory))
 
@@ -71,4 +90,4 @@ def delete(self):
         shutil.rmtree(self._path)
 
     def exists(self):
-        return os.path.isdir(self._path)
+        return os.path.isdir(self._path)

daliuge-engine/dlg/drop.py

@@ -1162,13 +1162,13 @@ def get_dir(self, dirname):
         parts = []
         if self._dlg_session:
             parts.append(".")
-            parts.append(self._dlg_session.sessionId)
         else:
             parts.append("/tmp/daliuge_tfiles")
         if dirname:
             parts.append(dirname)
 
         the_dir = os.path.abspath(os.path.normpath(os.path.join(*parts)))
+        logger.debug("Path used for drop: %s", the_dir)
         createDirIfMissing(the_dir)
         return the_dir
 
@@ -1561,6 +1561,7 @@ def exists(self):
             return any([c.exists() for c in self._children])
         return True
 
+
 # ===============================================================================
 # AppDROP classes follow
 # ===============================================================================

daliuge-engine/dlg/manager/session.py

@@ -164,6 +164,15 @@ def __init__(self, sessionId, nm=None):
         self._graphreprodata = None
         self._reprofinished = False
 
+        # create the session directory and change CWD
+        self._sessionDir = f"{utils.getDlgWorkDir()}/{sessionId}"
+        logger.debug("Creating session directory: %s", self._sessionDir)
+        createDirIfMissing(self._sessionDir)
+        os.chdir(self._sessionDir)
+
+        logger.debug("Updating ENV with SESSION_ID: %s", sessionId)
+        os.environ.update({"DLG_SESSION_ID": sessionId})
+
         class SessionFilter(logging.Filter):
             def __init__(self, sessionId):
                 self.sessionId = sessionId
@@ -177,10 +186,10 @@ def filter(self, record):
         fmt = logging.Formatter(fmt)
         fmt.converter = time.gmtime
 
-        logdir = utils.getDlgLogsDir()
-        if self._nm is not None:
-            logdir = self._nm.logdir
-        logfile = generateLogFileName(logdir, self.sessionId)
+        # logdir = utils.getDlgLogsDir()
+        # if self._nm is not None:
+        #     logdir = self._nm.logdir
+        logfile = generateLogFileName(self._sessionDir, self.sessionId)
         try:
             self.file_handler = logging.FileHandler(logfile)
             self.file_handler.setFormatter(fmt)
@@ -195,6 +204,10 @@ def filter(self, record):
     def sessionId(self):
         return self._sessionId
 
+    @property
+    def sessionDir(self):
+        return self._sessionDir
+
     @property
     def status(self):
         with self._statusLock:
@@ -227,8 +240,7 @@ def reprostatus(self, status):
             self._reprofinished = status
 
     def write_reprodata(self):
-        parts = [utils.getDlgLogsDir(), self._sessionId]
-        the_dir = os.path.abspath(os.path.normpath(os.path.join(*parts)))
+        the_dir = self._sessionDir
         createDirIfMissing(the_dir)
         the_path = os.path.join(the_dir, "reprodata.out")
         with open(the_path, "w+", encoding="utf-8") as file:

daliuge-engine/test/lifecycle/test_registry.py

@@ -22,13 +22,13 @@
 import os
 import sqlite3
 import unittest
+import tempfile
 
 from dlg.data.memory import InMemoryDROP
 from dlg.lifecycle.registry import RDBMSRegistry
 
 
-DBFILE = "testing_dlm.db"
-
+DBFILE = tempfile.mktemp()
 
 class TestRDBMSRegistry(unittest.TestCase):
     def setUp(self):

tools/xml2palette/xml2palette.py

@@ -1014,8 +1014,6 @@ def cleanString(text: str) -> str:
 def parseCasaDocs(dStr:str) -> dict:
     """
     Parse the special docstring for casatasks
-    """
-    """
     Extract the parameters from the casatask doc string.
 
     :param task: The casatask to derive the parameters from.