Flexible working directory

constrain/api/workflow.py

@@ -5,7 +5,8 @@
 """
 
 import glob
-import sys, logging, json, os, datetime
+import sys, logging, json, os, datetime, platform
+from pathlib import Path
 from typing import Union
 
 sys.path.append("./constrain")
@@ -37,7 +38,6 @@ def __init__(self, workflow, run_workflow_now=False):
         self.states = {}
         self.workflow_dict = {}
         self.running_sequence = []
-
         # checking workflow validity is handed over to the API method.
         if isinstance(workflow, str):
             self.load_workflow_json(workflow)
@@ -46,10 +46,68 @@ def __init__(self, workflow, run_workflow_now=False):
             self.workflow_dict = workflow
 
         self.load_states()
+        # change working dir
+        self.change_work_dir()
 
+        # run workflow now
         if run_workflow_now:
             self.run_workflow()
 
+    def change_work_dir(self) -> None:
+        # First, check if the workflow_dict has "working_dir"
+        if "working_dir" not in self.workflow_dict:
+            logging.info("No working_dir is specified")
+        else:
+            # Change the working path to the working_dir value in workflow_dict.
+            if not isinstance(self.workflow_dict["working_dir"], str):
+                # First, detect if the working dir is a valid string
+                logging.error("working directory specified is not a valid string.")
+            else:
+                # Then detect if the working dir provided is in Linux format or Windows Format.
+                if ("/" in self.workflow_dict["working_dir"]) and (
+                    "\\" not in self.workflow_dict["working_dir"]
+                ):
+                    # in Linux Format
+                    if (
+                        platform.system() == "Windows"
+                    ):  # convert it to the working platform if it is windows
+                        self.workflow_dict["working_dir"] = self.workflow_dict[
+                            "working_dir"
+                        ].replace("/", "\\")
+                    logging.info("the working dir provided is in Linux format.")
+                elif ("/" not in self.workflow_dict["working_dir"]) and (
+                    "\\" in self.workflow_dict["working_dir"]
+                ):
+                    # in windows format
+                    if (
+                        platform.system() == "Linux"
+                    ):  # convert it to the working platform if it is Linux
+                        self.workflow_dict["working_dir"] = self.workflow_dict[
+                            "working_dir"
+                        ].replace("\\", "/")
+                    logging.info("the working dir provided is in Win format.")
+
+                # change the working directory if it exists
+                if os.path.exists(self.workflow_dict["working_dir"]):
+                    os.chdir(self.workflow_dict["working_dir"])
+                    logging.info("Change current working path to the specified path.")
+                else:
+                    try:
+                        Path(self.workflow_dict["working_dir"]).mkdir(
+                            parents=True, exist_ok=True
+                        )
+                        logging.info(
+                            "working directory specified does not exist and create a new director."
+                        )
+                        os.chdir(self.workflow_dict["working_dir"])
+                    except Exception as e:
+                        # If an invalid escapse sequence string is specified. E.g. ".\tests\api\test"
+                        if e.winerror == 123:
+                            # The error is : OSError: [WinError 123] The filename, directory name, or volume label syntax is incorrect.
+                            logging.error(
+                                "The working directory specified is an invalid escape sequence string."
+                            )
+
     def validate(self, verbose: bool = False) -> bool:
         """function to be implemented to check for high level validity of the workflow definition"""
         workflow_schema = {
@@ -301,6 +359,7 @@ def create_workflow_engine(
         Returns:
             Union[None, WorkflowEngine]: Instantiated WorkflowEngine object if provided workflow is valid; None otherwise.
         """
+
         if (isinstance(workflow, str) and os.path.isfile(workflow)) or isinstance(
             workflow, dict
         ):

tests/api/data/flexible_calling_unit_test/verification_case_unit_test_Path.json

@@ -0,0 +1 @@
+{"workflow_name": "A minimal case", "meta": {"author": "ConStrain Team", "date": "03/23/2024", "version": "1.0", "description": "A minimal case to test if flexible call feature works"}, "imports": [], "states": {}}