run.py

"""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