#!/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
import platform
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 platform.system() == "Windows":
HOME = os.environ['USERPROFILE']
else:
HOME = os.environ['HOME']
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.
"""
prefix = "[show_and_return] "
# (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)
remove_bin = False # FIXME: True for debug only
# ^ Either True/False succeeds in bash, either fails in vscode
# with https://github.com/Poikilos/EnlivenMinetest/issues/616
if remove_bin:
cwd = os.path.join(cwd, "bin")
bin_rel = "./bin/"
if isinstance(cmd, list):
if cmd[0].startswith(bin_rel):
cmd[0] = "./" + cmd[0][len(bin_rel):]
else:
if cmd.startswith(bin_rel):
cmd = "./" + cmd[len(bin_rel):]
# pre_bin = "export LD_LIBRARY_PATH=/usr/lib64 " # doesn't fix #616
# Somehow #616 gone, even shell, force_shell, force_env False & no pre_bin
# Maybe restart computer or run `sudo ldconfig` but I did neither
# (unless installing a package triggered it--I installed lsb before the
# problem went away I think).
pre_bin = ""
force_shell = False
# ^ Either True/False succeeds in bash, either way LD_LIBRARY_PATH
# fails in vscode:
# https://github.com/Poikilos/EnlivenMinetest/issues/616
if not shell and force_shell:
# force_shell fails with shell=True (just hangs
# near `meta['bytes'] = err['source'].read(1)`,
# even if run-any is run from bash prompt manually)
if platform.system() == "Linux":
if isinstance(cmd, list):
# pre_bin = "LD_LIBRARY_PATH=/usr/lib64; export LD_LIBRARY_PATH; " # doesn't fix #616
# ^ adding "export " causes bad identifier error later in command
cmd = ["bash", "-c", "cd '" + cwd + "'; " + pre_bin + shlex.join(cmd)]
else:
cmd = ["bash", "-c", "cd '" + cwd + "'; " + pre_bin + cmd]
run_msg = prefix+"Running %s" % pformat(cmd)
run_msg += ' # shell=%s in "%s"' % (shell, cwd)
echo0(run_msg)
if cwd is not None:
os.chdir(cwd)
out = {
'bytes': None, # current line bytes
'string': "", # current line string
# ^ (same as bytes if Python 2 running)
'buffer': "", # cumulative buffer
}
good_path = (
"{HOME}/.rvm/gems/ruby-3.0.0/bin:"
"{HOME}/.rvm/gems/ruby-3.0.0@global/bin:"
"{HOME}/.rvm/rubies/ruby-3.0.0/bin:"
"{HOME}/.nvm/versions/node/v14.21.2/bin:"
"/usr/lib64/ccache:{HOME}/.cargo/bin:"
"/usr/local/bin:/usr/bin:/bin:"
"{HOME}/.local/bin:/usr/local/sbin:/usr/sbin:"
"{HOME}/.local/bin:"
"{HOME}/git/linux-preinstall/utilities:"
"{HOME}/git/linux-preinstall/utilities-developer:"
"{HOME}/git/linux-preinstall/utilities-server:"
"{HOME}/.rvm/bin:{HOME}/.local/bin:{HOME}/bin:"
"{HOME}/git/linux-preinstall/utilities:"
"{HOME}/git/linux-preinstall/utilities-server:"
"{HOME}/git/linux-preinstall/utilities-developer:"
"{HOME}/.local/bin:{HOME}/git/linux-preinstall/utilities:"
"{HOME}/git/linux-preinstall/utilities-developer:"
"{HOME}/git/linux-preinstall/utilities-server:"
"{HOME}/.rvm/bin"
"".format(HOME=HOME)
)
# ^ known good path in working bash that can run run-any
# os.environ['PATH'] = good_path # still doesn't fix issue #616
this_env = os.environ.copy()
force_env = False
if force_env and platform.system() == "Linux":
# For diagnosing issue #616
LD_LIBRARY_PATH = "/usr/lib64"
if "LD_LIBRARY_PATH" not in this_env:
print(prefix+"adding LD_LIBRARY_PATH=%s" % LD_LIBRARY_PATH)
this_env["LD_LIBRARY_PATH"] = LD_LIBRARY_PATH
elif LD_LIBRARY_PATH not in this_env['LD_LIBRARY_PATH']:
print(prefix+"LD_LIBRARY_PATH=%s"
% (os.environ['LD_LIBRARY_PATH']))
print(prefix+"appending %s to LD_LIBRARY_PATH" % LD_LIBRARY_PATH)
if not this_env["LD_LIBRARY_PATH"].endswith(os.pathsep):
this_env["LD_LIBRARY_PATH"] += os.pathsep
os.environ["LD_LIBRARY_PATH"] += os.pathsep
this_env["LD_LIBRARY_PATH"] += LD_LIBRARY_PATH
os.environ["LD_LIBRARY_PATH"] += LD_LIBRARY_PATH
print(prefix+"LD_LIBRARY_PATH=%s"
% (os.environ['LD_LIBRARY_PATH']))
else:
print(prefix+"detected %s in LD_LIBRARY_PATH=%s"
% (LD_LIBRARY_PATH, os.environ['LD_LIBRARY_PATH']))
enable_call = False
if enable_call: # FIXME: True for debug only--issue #616 (doesn't fix)
out['lines'] = []
# source, stream, and bytes members are discarded before return
# in non-dummy case (not enable_call) & not necessary for caller.
err = copy.deepcopy(out)
stream_metas = OrderedDict({
'out': out,
'err': err,
})
print(prefix+"calling (shell=%s): %s" % (shell, cmd),
file=sys.stderr)
code = subprocess.run(cmd, shell=shell, env=this_env)
return { # FIXME: for debug issue #616 only
'code': code,
'streams': stream_metas,
}
if shell:
cmd = pre_bin + cmd
print(prefix+"opening (shell=%s): %s" % (shell, cmd),
file=sys.stderr)
proc = subprocess.Popen(cmd, shell=shell, stderr=subprocess.PIPE,
stdout=subprocess.PIPE, cwd=cwd,
env=this_env)
code = None
# Do not wait for finish--start displaying output immediately
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'] += meta['buffer'].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,
}
issue_616_flag = "cannot open shared object file: No such file or directory"
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(os.path.realpath(path))
dir_name = os.path.basename(cwd)
if dir_name == "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:
echo0("Warning: not Minetest-like dir_name=%s" % dir_name)
else:
raise ValueError('missing "%s" (cwd="%s")--use absolute path if not in cwd'
% (path, os.getcwd()))
# 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.
else:
targetBaseDir = exeDir
if targetBaseDir == "":
targetBaseDir = os.getcwd()
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:
sys.stdout.write("\nOutputInspector:")
sys.stdout.flush()
count = 0
for line in all_lines:
inspector.addLine(line, enablePush)
count += 1
# NOTE: addLine adds all of the metadata!
if count < 1:
cmd_msg = cmd if isinstance(cmd, str) else shlex.join(cmd)
print(" (Output of `%s` had %s line(s) of output.)"
% (pformat(cmd_msg), count),
file=sys.stderr)
else:
print("", file=sys.stderr)
for info in inspector.getLineInfos():
print(info['all'], file=sys.stderr)
if issue_616_flag in info['all']:
print("- detected issue #616.")
print(" See <https://"
"github.com/Poikilos/EnlivenMinetest/issues/616>",
file=sys.stderr)
pass
# 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.
