Move logged_subprocess() from build to plugin_helers

Signed-off-by: Jeremy Fowers <[email protected]>

src/turnkeyml/common/build.py

@@ -18,7 +18,6 @@
 import sklearn.base
 import turnkeyml.common.exceptions as exp
 import turnkeyml.common.tf_helpers as tf_helpers
-import turnkeyml.run.plugin_helpers as plugin_helpers
 from turnkeyml.version import __version__ as turnkey_version
 
 
@@ -445,65 +444,6 @@ def flush(self):
         pass
 
 
-def logged_subprocess(
-    cmd: List[str],
-    cwd: str = os.getcwd(),
-    env: Optional[Dict] = None,
-    log_file_path: Optional[str] = None,
-    log_to_std_streams: bool = True,
-    log_to_file: bool = True,
-) -> None:
-    """
-    This function calls a subprocess and sends the logs to either a file, stdout/stderr, or both.
-
-    cmd             Command that will run o a sbprocess
-    cwd             Working directory from where the subprocess should run
-    env             Evironment to be used by the subprocess (useful for passing env vars)
-    log_file_path   Where logs will be stored
-    log_to_file     Whether or not to store the subprocess's stdout/stderr into a file
-    log_to_std      Whether or not to print subprocess's stdout/stderr to the screen
-    """
-    if env is None:
-        env = os.environ.copy()
-    if log_to_file and log_file_path is None:
-        raise ValueError("log_file_path must be set when log_to_file is True")
-
-    log_stdout = ""
-    log_stderr = ""
-    try:
-        proc = subprocess.run(
-            cmd,
-            check=True,
-            env=env,
-            capture_output=True,
-            cwd=cwd,
-        )
-    except Exception as e:  # pylint: disable=broad-except
-        log_stdout = e.stdout.decode("utf-8")  # pylint: disable=no-member
-        log_stderr = e.stderr.decode("utf-8")  # pylint: disable=no-member
-        raise plugin_helpers.CondaError(
-            f"Exception {e} encountered, \n\nstdout was: "
-            f"\n{log_stdout}\n\n and stderr was: \n{log_stderr}"
-        )
-    else:
-        log_stdout = proc.stdout.decode("utf-8")
-        log_stderr = proc.stderr.decode("utf-8")
-    finally:
-        if log_to_std_streams:
-            # Print log to stdout
-            # This might be useful when this subprocess is being logged externally
-            print(log_stdout, file=sys.stdout)
-            print(log_stderr, file=sys.stdout)
-        if log_to_file:
-            log = f"{log_stdout}\n{log_stderr}"
-            with open(
-                log_file_path,
-                "w",
-                encoding="utf-8",
-            ) as f:
-                f.write(log)
-
-
 def get_system_info():
     os_type = platform.system()
     info_dict = {}

src/turnkeyml/run/onnxrt/execute.py

@@ -8,7 +8,6 @@
 import json
 from statistics import mean
 import platform
-import turnkeyml.common.build as build
 import turnkeyml.run.plugin_helpers as plugin_helpers
 
 ORT_VERSION = "1.15.1"
@@ -84,7 +83,7 @@ def execute_benchmark(
     ]
 
     # Execute command and log stdout/stderr
-    build.logged_subprocess(
+    plugin_helpers.logged_subprocess(
         cmd=cmd,
         cwd=os.path.dirname(output_dir),
         log_to_std_streams=False,

src/turnkeyml/run/plugin_helpers.py

@@ -1,6 +1,8 @@
 import subprocess
 import logging
 import os
+import sys
+from typing import List, Optional, Dict
 
 TIMEOUT = 900
 
@@ -72,3 +74,62 @@ def get_python_path(conda_env_name):
             f"An error occurred while getting Python path for {conda_env_name} environment"
             f"{e.stderr.decode()}"
         )
+
+
+def logged_subprocess(
+    cmd: List[str],
+    cwd: str = os.getcwd(),
+    env: Optional[Dict] = None,
+    log_file_path: Optional[str] = None,
+    log_to_std_streams: bool = True,
+    log_to_file: bool = True,
+) -> None:
+    """
+    This function calls a subprocess and sends the logs to either a file, stdout/stderr, or both.
+
+    cmd             Command that will run o a sbprocess
+    cwd             Working directory from where the subprocess should run
+    env             Evironment to be used by the subprocess (useful for passing env vars)
+    log_file_path   Where logs will be stored
+    log_to_file     Whether or not to store the subprocess's stdout/stderr into a file
+    log_to_std      Whether or not to print subprocess's stdout/stderr to the screen
+    """
+    if env is None:
+        env = os.environ.copy()
+    if log_to_file and log_file_path is None:
+        raise ValueError("log_file_path must be set when log_to_file is True")
+
+    log_stdout = ""
+    log_stderr = ""
+    try:
+        proc = subprocess.run(
+            cmd,
+            check=True,
+            env=env,
+            capture_output=True,
+            cwd=cwd,
+        )
+    except Exception as e:  # pylint: disable=broad-except
+        log_stdout = e.stdout.decode("utf-8")  # pylint: disable=no-member
+        log_stderr = e.stderr.decode("utf-8")  # pylint: disable=no-member
+        raise CondaError(
+            f"Exception {e} encountered, \n\nstdout was: "
+            f"\n{log_stdout}\n\n and stderr was: \n{log_stderr}"
+        )
+    else:
+        log_stdout = proc.stdout.decode("utf-8")
+        log_stderr = proc.stderr.decode("utf-8")
+    finally:
+        if log_to_std_streams:
+            # Print log to stdout
+            # This might be useful when this subprocess is being logged externally
+            print(log_stdout, file=sys.stdout)
+            print(log_stderr, file=sys.stdout)
+        if log_to_file:
+            log = f"{log_stdout}\n{log_stderr}"
+            with open(
+                log_file_path,
+                "w",
+                encoding="utf-8",
+            ) as f:
+                f.write(log)

test/unit.py

@@ -101,8 +101,8 @@ def test_003_subprocess_logger(self):
         traceback_error_cmd = f"raise ValueError('{traceback_error_msg}')"
 
         # Perform basic test (no exceptions inside logger)
-        cmd = ["python","-c",f"import sys\n{inside_stdout_cmd}\n{inside_sterr_cmd}"]
-        build.logged_subprocess(cmd=cmd, log_file_path=logfile_path)
+        cmd = ["python", "-c", f"import sys\n{inside_stdout_cmd}\n{inside_sterr_cmd}"]
+        plugin_helpers.logged_subprocess(cmd=cmd, log_file_path=logfile_path)
 
         # Make sure we captured everything we intended to capture
         with open(logfile_path, "r", encoding="utf-8") as file:
@@ -111,9 +111,13 @@ def test_003_subprocess_logger(self):
         assert inside_sterr_msg in log_contents
 
         # Perform test with exceptions inside the logger
-        cmd = ["python","-c",f"import sys\n{inside_stdout_cmd}\n{inside_sterr_cmd}\n{traceback_error_cmd}"]
+        cmd = [
+            "python",
+            "-c",
+            f"import sys\n{inside_stdout_cmd}\n{inside_sterr_cmd}\n{traceback_error_cmd}",
+        ]
         with self.assertRaises(plugin_helpers.CondaError):
-            build.logged_subprocess(cmd=cmd, log_file_path=logfile_path)
+            plugin_helpers.logged_subprocess(cmd=cmd, log_file_path=logfile_path)
 
         # Make sure we captured everything we intended to capture
         with open(logfile_path, "r", encoding="utf-8") as file:
@@ -126,16 +130,24 @@ def test_003_subprocess_logger(self):
         subprocess_env = os.environ.copy()
         expected_env_var_value = "Expected Value"
         subprocess_env["TEST_ENV_VAR"] = expected_env_var_value
-        cmd = ["python","-c",f'import os\nprint(os.environ["TEST_ENV_VAR"])']
-        build.logged_subprocess(cmd=cmd, log_file_path=logfile_path, env=subprocess_env)
+        cmd = ["python", "-c", f'import os\nprint(os.environ["TEST_ENV_VAR"])']
+        plugin_helpers.logged_subprocess(
+            cmd=cmd, log_file_path=logfile_path, env=subprocess_env
+        )
         with open(logfile_path, "r", encoding="utf-8") as file:
             log_contents = file.read()
         assert expected_env_var_value in log_contents
 
         # Test log_to_std_streams
-        cmd = ["python","-c",f'print("{outside_stdout_msg}")\nprint("{outside_stderr_msg}")']
+        cmd = [
+            "python",
+            "-c",
+            f'print("{outside_stdout_msg}")\nprint("{outside_stderr_msg}")',
+        ]
         with build.Logger("", logfile_path):
-            build.logged_subprocess(cmd=cmd, log_to_std_streams=True, log_to_file=False)
+            plugin_helpers.logged_subprocess(
+                cmd=cmd, log_to_std_streams=True, log_to_file=False
+            )
         with open(logfile_path, "r", encoding="utf-8") as file:
             log_contents = file.read()
         assert outside_stdout_msg in log_contents