Cache: Add working directory parameter

Author: jan-janssen

executorlib/cache/executor.py

@@ -1,4 +1,5 @@
 import os
+from typing import Optional
 
 from executorlib.base.executor import ExecutorBase
 from executorlib.cache.shared import execute_in_subprocess, execute_tasks_h5
@@ -11,6 +12,7 @@ def __init__(
         cache_directory: str = "cache",
         execute_function: callable = execute_in_subprocess,
         cores_per_worker: int = 1,
+        cwd: Optional[str] = None,
     ):
         """
         Initialize the FileExecutor.
@@ -19,6 +21,7 @@ def __init__(
             cache_directory (str, optional): The directory to store cache files. Defaults to "cache".
             execute_function (callable, optional): The function to execute tasks. Defaults to execute_in_subprocess.
             cores_per_worker (int, optional): The number of CPU cores per worker. Defaults to 1.
+            cwd (str/None): current working directory where the parallel python task is executed
         """
         super().__init__()
         cache_directory_path = os.path.abspath(cache_directory)
@@ -31,6 +34,7 @@ def __init__(
                     "execute_function": execute_function,
                     "cache_directory": cache_directory_path,
                     "cores_per_worker": cores_per_worker,
+                    "cwd": cwd,
                 },
             )
         )

executorlib/cache/shared.py

@@ -4,7 +4,7 @@
 import subprocess
 import sys
 from concurrent.futures import Future
-from typing import Tuple
+from typing import Tuple, Optional
 
 from executorlib.standalone.command import get_command_path
 from executorlib.standalone.hdf import dump, get_output
@@ -48,14 +48,15 @@ def done(self) -> bool:
 
 
 def execute_in_subprocess(
-    command: list, task_dependent_lst: list = []
+    command: list, task_dependent_lst: list = [], cwd: Optional[str] = None,
 ) -> subprocess.Popen:
     """
     Execute a command in a subprocess.
 
     Args:
         command (list): The command to be executed.
-        task_dependent_lst (list, optional): A list of subprocesses that the current subprocess depends on. Defaults to [].
+        task_dependent_lst (list): A list of subprocesses that the current subprocess depends on. Defaults to [].
+        cwd (str/None): current working directory where the parallel python task is executed
 
     Returns:
         subprocess.Popen: The subprocess object.
@@ -65,14 +66,15 @@ def execute_in_subprocess(
         task_dependent_lst = [
             task for task in task_dependent_lst if task.poll() is None
         ]
-    return subprocess.Popen(command, universal_newlines=True)
+    return subprocess.Popen(command, universal_newlines=True, cwd=cwd)
 
 
 def execute_tasks_h5(
     future_queue: queue.Queue,
     cache_directory: str,
     cores_per_worker: int,
     execute_function: callable,
+    cwd: Optional[str],
 ) -> None:
     """
     Execute tasks stored in a queue using HDF5 files.
@@ -82,6 +84,7 @@ def execute_tasks_h5(
         cache_directory (str): The directory to store the HDF5 files.
         cores_per_worker (int): The number of cores per worker.
         execute_function (callable): The function to execute the tasks.
+        cwd (str/None): current working directory where the parallel python task is executed
 
     Returns:
         None
@@ -123,6 +126,7 @@ def execute_tasks_h5(
                         task_dependent_lst=[
                             process_dict[k] for k in future_wait_key_lst
                         ],
+                        cwd=cwd,
                     )
                 file_name_dict[task_key] = os.path.join(
                     cache_directory, task_key + ".h5out"

tests/test_cache_executor_serial.py

@@ -19,6 +19,10 @@ def my_funct(a, b):
     return a + b
 
 
+def list_files_in_working_directory():
+    return os.listdir(os.getcwd())
+
+
 @unittest.skipIf(
     skip_h5io_test, "h5io is not installed, so the h5io tests are skipped."
 )
@@ -38,6 +42,12 @@ def test_executor_dependence_mixed(self):
             self.assertEqual(fs2.result(), 4)
             self.assertTrue(fs2.done())
 
+    def test_executor_working_directory(self):
+        cwd = os.path.join(os.path.dirname(__file__), "executables")
+        with FileExecutor(cwd=cwd) as exe:
+            fs1 = exe.submit(list_files_in_working_directory)
+            self.assertEqual(fs1.result(), os.listdir(cwd))
+
     def test_executor_function(self):
         fs1 = Future()
         q = Queue()