"""Utility for subprocess calls.""" import logging import os import subprocess import time def run(*args, counter=None, **kwargs): """ Run a subprocess and log the call. Convenience function to call `subprocess.run` and provide some metadata about the call. Parameters ---------- args : tuple passed on to subprocess.run(*args). For example ("ls -l") or (["ls", "-l"]) counter : TYPE, optional list with one integer, starting from [0]. This is (a Python hack) to count the number of calls of this function, so the different calls of subprocess.run are easier to follow in the log files. kwargs : dict keyword-value arguments to be passed to subprocess.run. For example, `shell=True`. Returns ------- process : subprocess.CompletedProcess CompletedProcess object as returned by `subprocess.run` . Examples -------- >>> run(['date', '+%Y-%m-%d']) ##-03 Starting subprocess.run(['date', '+%Y-%m-%d']) with options ##-03 getcwd=/Users/fangohr/git/mpsd-software-environments ##-03 COMMAND=date +%Y-%m-%d 2023-05-30 ##-03 Completed in 0.0054s. ##-03 CompletedProcess(args=['date', '+%Y-%m-%d'], returncode=0) >>> run(['date +%Y-%m-%d'], shell=True) ##-04 Starting subprocess.run(['date +%Y-%m-%d']) with options shell=True ##-04 getcwd=/Users/fangohr/git/mpsd-software-environments ##-04 COMMAND=date +%Y-%m-%d 2023-05-30 ##-04 Completed in 0.0069s. ##-04 CompletedProcess(args=['date +%Y-%m-%d'], returncode=0) """ # token is printed in front of every meta-data line - useful for # searching the logs. Starts with "##-00", then "##-01", ... if counter is None: counter = [0] token = f"##-{counter[0]:02d}" counter[0] += 1 # increase counter # make command nicely readable: ["ls", "-l"] -> "ls -l" assert isinstance(args, tuple) assert len(args) == 1 arg = args[0] # either args is a tuple containing a string | Example: ('ls -1',) if isinstance(arg, str): command = arg # or we have a tuple containing a list of strings. # Example: (['ls', '-1'],) elif isinstance(arg, list): command = " ".join(arg) else: # we do not expect this to happen raise NotImplementedError(f"arg={arg}, args={args}") # Translate `capture_output' for compatibility with older Python if kwargs.pop("capture_output", False): if kwargs.get("stdout") is not None or kwargs.get("stderr") is not None: raise ValueError( "stdout and stderr arguments may not be used with capture_output." ) kwargs["stdout"] = subprocess.PIPE kwargs["stderr"] = subprocess.PIPE # make options (such as `shell=True`) nicely readable options = ", ".join([f"{key}={value}" for key, value in kwargs.items()]) # provide information about upcoming subprocess.run call logging.debug( f"{token} Starting subprocess.run('{command}') with options {options}" ) logging.debug(f"""{token} getcwd={os.getcwd()}""") logging.debug(f"""{token} subprocess.run("{arg}")""") time_start = time.time() process = subprocess.run(*args, **kwargs) execution_time = time.time() - time_start logging.debug(f"{token} process={process}") logging.debug(f"{token} Completed in {execution_time:.4f}s.") logging.debug(f"{token}") # near-empty line to make reading logs easier return process