rolled "base.execute" into "read_and_display" including error common handling performed in Environment.execute

Author: tclose

pydra/environments/base.py

@@ -200,52 +200,46 @@ def map_path(fileset: os.PathLike | FileSet) -> Path:
         return bindings, values
 
 
-def execute(cmd, strip=False):
-    """
-    Run the event loop with coroutine.
-
-    Uses :func:`read_and_display_async` unless a loop is
-    already running, in which case :func:`read_and_display`
-    is used.
+def read_and_display(
+    *cmd: str, strip: bool = False, hide_display: bool = False
+) -> dict[str, int | str]:
+    """Capture a process' standard output.
 
     Parameters
     ----------
-    cmd : :obj:`list` or :obj:`tuple`
-        The command line to be executed.
-    strip : :obj:`bool`
-        TODO
-
-    """
-    rc, stdout, stderr = read_and_display(*cmd, strip=strip)
-    """
-    loop = get_open_loop()
-    if loop.is_running():
-        rc, stdout, stderr = read_and_display(*cmd, strip=strip)
-    else:
-        rc, stdout, stderr = loop.run_until_complete(
-            read_and_display_async(*cmd, strip=strip)
-        )
+    cmd : str
+        The command to execute, as a list of strings.
+    strip : bool, optional
+        If True, the output will be stripped of leading and trailing whitespace.
+    hide_display : bool, optional
+        If True, the output will not be displayed.
+
+    Returns
+    -------
+        dict[str, Any]
+            A dictionary containing the return code, standard output, and standard error.
+
+    Raises
+    ------
+        RuntimeError
+            If the return code is not 0, a RuntimeError is raised with a formatted
+            error message.
     """
-    return rc, stdout, stderr
-
-
-def read_and_display(*cmd, strip=False, hide_display=False):
-    """Capture a process' standard output."""
     try:
         process = sp.run(cmd, stdout=sp.PIPE, stderr=sp.PIPE)
     except Exception:
         # TODO editing some tracing?
         raise
 
+    stdout = process.stdout.decode("utf-8")
     if strip:
-        return (
-            process.returncode,
-            process.stdout.decode("utf-8").strip(),
-            process.stderr.decode("utf-8"),
-        )
-    else:
-        return (
-            process.returncode,
-            process.stdout.decode("utf-8"),
-            process.stderr.decode("utf-8"),
-        )
+        stdout = stdout.strip()
+    stderr = process.stderr.decode("utf-8")
+    if process.returncode:
+        msg = f"Error executing command {cmd} (code: {process.returncode}):"
+        if stdout:
+            msg += "\n\nstderr:\n" + stderr
+        if stdout:
+            msg += "\n\nstdout:\n" + stdout
+        raise RuntimeError(msg)
+    return {"return_code": process.returncode, "stdout": stdout, "stderr": stderr}

pydra/environments/docker.py

@@ -28,19 +28,11 @@ def execute(self, job: "Job[shell.Task]") -> dict[str, ty.Any]:
             ).split()
         )
         docker_args.extend(["-w", f"{self.root}{job.cache_dir}"])
-        keys = ["return_code", "stdout", "stderr"]
 
         job.cache_dir.mkdir(exist_ok=True)
-        values = base.execute(
-            docker_args + [docker_img] + job.task._command_args(values=values),
+        return base.read_and_display(
+            *(docker_args + [docker_img] + job.task._command_args(values=values)),
         )
-        output = dict(zip(keys, values))
-        if output["return_code"]:
-            if output["stderr"]:
-                raise RuntimeError(output["stderr"])
-            else:
-                raise RuntimeError(output["stdout"])
-        return output
 
 
 # Alias so it can be referred to as docker.Environment

pydra/environments/native.py

@@ -15,18 +15,8 @@ class Native(base.Environment):
     """
 
     def execute(self, job: "Job[shell.Task]") -> dict[str, ty.Any]:
-        keys = ["return_code", "stdout", "stderr"]
         cmd_args = job.task._command_args(values=job.inputs)
-        values = base.execute(cmd_args)
-        output = dict(zip(keys, values))
-        if output["return_code"]:
-            msg = f"Error running '{job.name}' job with {cmd_args}:"
-            if output["stderr"]:
-                msg += "\n\nstderr:\n" + output["stderr"]
-            if output["stdout"]:
-                msg += "\n\nstdout:\n" + output["stdout"]
-            raise RuntimeError(msg)
-        return output
+        return base.read_and_display(cmd_args)
 
 
 # Alias so it can be referred to as native.Environment

pydra/environments/singularity.py

@@ -31,21 +31,12 @@ def execute(self, job: "Job[shell.Task]") -> dict[str, ty.Any]:
         singularity_args.extend(
             ["--pwd", f"{self.root.rstrip('/')}{job.cache_dir.absolute()}"]
         )
-        keys = ["return_code", "stdout", "stderr"]
-
         job.cache_dir.mkdir(exist_ok=True)
-        values = base.execute(
+        return base.read_and_display(
             singularity_args
             + [singularity_img]
             + job.task._command_args(values=values),
         )
-        output = dict(zip(keys, values))
-        if output["return_code"]:
-            if output["stderr"]:
-                raise RuntimeError(output["stderr"])
-            else:
-                raise RuntimeError(output["stdout"])
-        return output
 
 
 # Alias so it can be referred to as singularity.Environment