EnlivenMinetest/utilities/run-any

#!/usr/bin/env python
"""
Run any program and show the output. The purpose is to allow VSCode's
python runner to run a minetest binary (such as if you are using VSCode
to edit Lua but want to run Minetest to test the Lua).

Usage:
run-any <minetest_executable_path>
"""
from __future__ import print_function
import os
import sys
import subprocess
import copy
import shlex
from pprint import pformat
from collections import OrderedDict

SCRIPTS_DIR = os.path.dirname(os.path.realpath(__file__))
REPO_DIR = os.path.dirname(SCRIPTS_DIR)
REPOS_DIR = os.path.dirname(REPO_DIR)
for try_dirname in ["outputinspector-python", "outputinspector"]:
    try_path = os.path.join(REPOS_DIR, try_dirname)
    if os.path.isfile(os.path.join(try_path, "outputinspector",
                                    "__init__.py")):
        # ^ Yes, it is in outputinspector/outputinspector/
        #   If in same git dir as REPO_DIR rather than installed as a module
        sys.path.insert(0, try_path)
        break

if sys.version_info.major < 3:
    ModuleNotFoundError = ImportError
    FileNotFoundError = IOError
    input = raw_input

inspector = None
try:
    import outputinspector
    print("OutputInspector...FOUND.", file=sys.stderr)
    from outputinspector import (
        OutputInspector,
        ROLE_COLLECTED_FILE,
        ROLE_ROW,
        ROLE_COL,
        ROLE_LOWER,
        ROLE_COLLECTED_LINE,
        ROLE_DETAILS,
    )
    inspector = OutputInspector()
except ModuleNotFoundError as ex:
    print(ex, file=sys.stderr)
    print("INFO: Install outputinspector or clone to %s"
          " for clickable Lua traceback lines"
          % REPOS_DIR, file=sys.stderr)

def echo0(*args):
    print(*args, file=sys.stderr)
    return True


def usage():
    echo0(__doc__)


def show_and_return(cmd, enable_collect=False, cwd=None, shell=False):
    """Show the output of a process while it is running & get return code.

    Warning: Minetest doesn't start flushing data until the server
    starts. Output from mainmenu is flushed around when server starts
    (or when user exits main menu, whichever comes first). This behavior
    is the same whether or not shell is True.

    If running minetest, set cmd to [minetest_path, "--logfile", ""]
    and set cwd to minetest directory (not "bin"!).

    Args:
        cmd (Union[list[string], string]): The command to run, either as
            list of args where first is command and no quotes are
            necessary, or space-separated string formatted the same way
            except with quotes where necessary (if an arg
            contains spaces).
            - If is *not* str and shell is True, cmd is converted to
            str.
        enable_collect (Optional[boolean]): Enable collecting lines
            in the 'lines' key of the return.
        cwd (Optional[string]): Set a new current working dir.
        shell (boolean): If True, run a shell (processes globs and
            shell options that occur within cmd).

    Returns:
        dict: 'code' is return code of the command.
    """
    # (See
    # <https://cyberciti.biz/faq/python-run-external-command-and-get-output/>)
    if shell not in [None, True, False]:
        raise ValueError("Expected None, True, or False for shell but got %s"
                         % pformat(shell))
    if cwd is None:
        cwd = os.getcwd()
    if shell and not hasattr(cmd, "split"):
        # shell cannot correctly utilize a list/tuple (only first
        #   element is used!) so join as string to use all arguments:
        cmd = shlex.join(cmd)
    echo0("Running %s" % pformat(cmd))
    proc = subprocess.Popen(cmd, shell=shell, stderr=subprocess.PIPE,
                            stdout=subprocess.PIPE, cwd=cwd)
    code = None
    # Do not wait for finish--start displaying output immediately
    out = {
        'bytes': None,  # current line bytes
        'string': "",  # current line string (same as bytes if Python 2 running)
        'buffer': "",  # cumulative buffer
    }
    if enable_collect:
        out['lines'] = []  # already-shown lines
    err = copy.deepcopy(out)  # sets err['lines'] if enable_collect
    if enable_collect:
        if id(err['lines']) == id(out['lines']):
            raise RuntimeError(
                "deepcopy filed. lines are same in both streams"
            )
    out['source'] = proc.stdout
    err['source'] = proc.stderr
    out['stream'] = sys.stdout
    err['stream'] = sys.stderr
    stream_metas = OrderedDict({
        'out': out,
        'err': err,
    })

    while True:
        for name, meta in stream_metas.items():
            # meta['bytes'] = err['source'].read(1)
            # ^ scrambles letters (mixes stderr and stdout) somehow, so:
            if name == "out":
                meta['bytes'] = proc.stdout.read(1)
            elif name == "err":
                meta['bytes'] = proc.stderr.read(1)
            meta['string'] = meta['bytes']
            if (meta['string'] is not None) and (sys.version_info.major >= 3):
                meta['string'] = meta['bytes'].decode("utf-8")
            meta['buffer'] += meta['string']

        code = proc.poll()  # None if command isn't finished
        if out['string'] == '' and err['string'] == '' and code is not None:
            # If no more to write *and* there is a return code (finished)
            for name, meta in stream_metas.items():
                if len(meta['buffer']) > 0:
                    if name == "out":
                        print(meta['buffer'].rstrip("\n\r"))
                    else:
                        echo0(meta['buffer'].rstrip("\n\r"))
                    meta['lines'] += this_chunk.split("\n")
            break

        for _, meta in stream_metas.items():
            # Write out['string'] to stdout then err['string'] to stderr
            if meta['buffer'] == '':
                continue
            lastI = meta['buffer'].rfind("\n")
            if lastI >= 0:
                # Only show finished lines (avoid mixing stdout and
                #   stderr on the same line).
                if code is not None:
                    # The program ended, so show everything.
                    lastI = len(meta['buffer']) - 1
                    # ^ -1 since inclusive (+1 below)

                this_chunk = meta['buffer'][:lastI+1]


                if enable_collect:
                    # Don't use append (since split always returns a list
                    #   even if 1-long).
                    meta['lines'] += this_chunk.split("\n")
                meta['stream'].write(this_chunk)
                meta['stream'].flush()
                meta['buffer'] = meta['buffer'][lastI+1:]
            # else:
            #     echo0("Incomplete line:%s" % meta['buffer'])
    for _, meta in stream_metas.items():
        del meta['source']
        del meta['stream']
        del meta['bytes']
        # del meta['string']
        if len(meta['string']) > 0:
            raise NotImplementedError(
                "The remaining %s wasn't processed: %s"
                % (_, pformat(meta['string']))
            )
    return {
        'code': code,
        'streams': stream_metas,
    }


def main():
    enablePush = True  # collect the line in the inspector ui right away
    if len(sys.argv) < 2:
        usage()
        echo0("Error: You didn't specify what program to run.")
        return 1
    # subprocess.call(sys.argv[1])
    path = sys.argv[1]
    if path.lower().endswith(".txt"):
        raise ValueError("expected an executable!")
    cwd = None
    if os.path.isfile(path):
        cwd = os.path.dirname(path)
        if os.path.basename(cwd) == "bin":
            # Minetest must run from minetest not from minetest/bin
            # (especially when minetest is a script in minetest.org
            # builds)
            cwd = os.path.dirname(cwd)
    # else:
    #     cwd = os.dirname(hierosoft.which(path))  # TODO: uncomment this case?
    basename = os.path.basename(path)
    project_name = basename.replace("server", "")
    cmd = path
    exeDir = os.path.dirname(path)
    if project_name in ["minetest", "finetest", "trolltest", "multicraft"]:
        cmd = [path, "--logfile", ""]
        targetBaseDir = os.path.dirname(exeDir)
        # minetest/ not minetest/bin/ is the root of paths in tracebacks
        # Do not write debug.txt to cwd, since Python code will read
        #   stdout and stderr.
        echo0("Running %s" % shlex.join(cmd))
    else:
        targetBaseDir = exeDir
        echo0("Running %s" % path)
    if inspector:
        OutputInspector.addRoot(targetBaseDir)
    enable_collect = True
    results = show_and_return(
        cmd,
        enable_collect=enable_collect,
        cwd=cwd,
        # TODO: silent=not inspector,  # If there is no outputinspector, show in realtime
    )
    # ^ TODO: push to outputinspector in realtime
    if enable_collect:
        # echo0("\nLINES:")
        # for line in results['streams']['out']['lines']:
        #     print(line)
        # for line in results['streams']['err']['lines']:
        #     echo0(line)
        out = results['streams']['out']
        err = results['streams']['err']
    # else 'lines' lists will not be in the streams!

    if results['code'] == 0:
        echo0("No errors (return code 0)")
        # return results['code']

    err_index = len(out['lines'])
    all_lines = out['lines'] + err['lines']


    if inspector:
        print("\nOutputInspector:")
        for line in all_lines:
            inspector.addLine(line, enablePush)
        # NOTE: addLine adds all of the metadata!
        for info in inspector.getLineInfos():
            print(info['all'], file=sys.stderr)

        # raise SyntaxError(
        #     "{} line(s)".format(len(all_lines))+"\n".join(all_lines)
        # )
    # else output should already have been shown in realtime.
    return results['code']


if __name__ == "__main__":
    code = main()
    #sys.exit(code)
    sys.exit(0)  # Don't confuse VSCode. If nonzero, a popup will point here.