mirror of
https://github.com/ynput/ayon-core.git
synced 2025-12-24 21:04:40 +01:00
248336bb0d
* implemented 'is_func_signature_supported' function * 'WeakMethod' can be imported from 'python_2_comp' all the time * simplified events logic for callback registration * modified docstrings in publish lib * removed unused imports * fixed 'run_openpype_process' docstring
379 lines
11 KiB
Python
379 lines
11 KiB
Python
import os
|
|
import sys
|
|
import subprocess
|
|
import platform
|
|
import json
|
|
import tempfile
|
|
|
|
from .log import Logger
|
|
from .vendor_bin_utils import find_executable
|
|
|
|
from .openpype_version import is_running_from_build
|
|
|
|
# MSDN process creation flag (Windows only)
|
|
CREATE_NO_WINDOW = 0x08000000
|
|
|
|
|
|
def execute(args,
|
|
silent=False,
|
|
cwd=None,
|
|
env=None,
|
|
shell=None):
|
|
"""Execute command as process.
|
|
|
|
This will execute given command as process, monitor its output
|
|
and log it appropriately.
|
|
|
|
.. seealso::
|
|
|
|
:mod:`subprocess` module in Python.
|
|
|
|
Args:
|
|
args (list): list of arguments passed to process.
|
|
silent (bool): control output of executed process.
|
|
cwd (str): current working directory for process.
|
|
env (dict): environment variables for process.
|
|
shell (bool): use shell to execute, default is no.
|
|
|
|
Returns:
|
|
int: return code of process
|
|
|
|
"""
|
|
|
|
log_levels = ['DEBUG:', 'INFO:', 'ERROR:', 'WARNING:', 'CRITICAL:']
|
|
|
|
log = Logger.get_logger('execute')
|
|
log.info("Executing ({})".format(" ".join(args)))
|
|
popen = subprocess.Popen(
|
|
args,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
universal_newlines=True,
|
|
bufsize=1,
|
|
cwd=cwd,
|
|
env=env or os.environ,
|
|
shell=shell
|
|
)
|
|
|
|
# Blocks until finished
|
|
while True:
|
|
line = popen.stdout.readline()
|
|
if line == '':
|
|
break
|
|
if silent:
|
|
continue
|
|
line_test = False
|
|
for test_string in log_levels:
|
|
if line.startswith(test_string):
|
|
line_test = True
|
|
break
|
|
if not line_test:
|
|
print(line[:-1])
|
|
|
|
log.info("Execution is finishing up ...")
|
|
|
|
popen.wait()
|
|
return popen.returncode
|
|
|
|
|
|
def run_subprocess(*args, **kwargs):
|
|
"""Convenience method for getting output errors for subprocess.
|
|
|
|
Output logged when process finish.
|
|
|
|
Entered arguments and keyword arguments are passed to subprocess Popen.
|
|
|
|
On windows are 'creationflags' filled with flags that should cause ignore
|
|
creation of new window.
|
|
|
|
Args:
|
|
*args: Variable length argument list passed to Popen.
|
|
**kwargs : Arbitrary keyword arguments passed to Popen. Is possible to
|
|
pass `logging.Logger` object under "logger" to use custom logger
|
|
for output.
|
|
|
|
Returns:
|
|
str: Full output of subprocess concatenated stdout and stderr.
|
|
|
|
Raises:
|
|
RuntimeError: Exception is raised if process finished with nonzero
|
|
return code.
|
|
"""
|
|
|
|
# Modify creation flags on windows to hide console window if in UI mode
|
|
if (
|
|
platform.system().lower() == "windows"
|
|
and "creationflags" not in kwargs
|
|
# shell=True already tries to hide the console window
|
|
# and passing these creationflags then shows the window again
|
|
# so we avoid it for shell=True cases
|
|
and kwargs.get("shell") is not True
|
|
):
|
|
kwargs["creationflags"] = (
|
|
subprocess.CREATE_NEW_PROCESS_GROUP
|
|
| getattr(subprocess, "DETACHED_PROCESS", 0)
|
|
| getattr(subprocess, "CREATE_NO_WINDOW", 0)
|
|
)
|
|
|
|
# Get environents from kwarg or use current process environments if were
|
|
# not passed.
|
|
env = kwargs.get("env") or os.environ
|
|
# Make sure environment contains only strings
|
|
filtered_env = {str(k): str(v) for k, v in env.items()}
|
|
|
|
# Use lib's logger if was not passed with kwargs.
|
|
logger = kwargs.pop("logger", None)
|
|
if logger is None:
|
|
logger = Logger.get_logger("run_subprocess")
|
|
|
|
# set overrides
|
|
kwargs["stdout"] = kwargs.get("stdout", subprocess.PIPE)
|
|
kwargs["stderr"] = kwargs.get("stderr", subprocess.PIPE)
|
|
kwargs["stdin"] = kwargs.get("stdin", subprocess.PIPE)
|
|
kwargs["env"] = filtered_env
|
|
|
|
proc = subprocess.Popen(*args, **kwargs)
|
|
|
|
full_output = ""
|
|
_stdout, _stderr = proc.communicate()
|
|
if _stdout:
|
|
_stdout = _stdout.decode("utf-8", errors="backslashreplace")
|
|
full_output += _stdout
|
|
logger.debug(_stdout)
|
|
|
|
if _stderr:
|
|
_stderr = _stderr.decode("utf-8", errors="backslashreplace")
|
|
# Add additional line break if output already contains stdout
|
|
if full_output:
|
|
full_output += "\n"
|
|
full_output += _stderr
|
|
logger.info(_stderr)
|
|
|
|
if proc.returncode != 0:
|
|
exc_msg = "Executing arguments was not successful: \"{}\"".format(args)
|
|
if _stdout:
|
|
exc_msg += "\n\nOutput:\n{}".format(_stdout)
|
|
|
|
if _stderr:
|
|
exc_msg += "Error:\n{}".format(_stderr)
|
|
|
|
raise RuntimeError(exc_msg)
|
|
|
|
return full_output
|
|
|
|
|
|
def clean_envs_for_openpype_process(env=None):
|
|
"""Modify environments that may affect OpenPype process.
|
|
|
|
Main reason to implement this function is to pop PYTHONPATH which may be
|
|
affected by in-host environments.
|
|
"""
|
|
if env is None:
|
|
env = os.environ
|
|
|
|
# Exclude some environment variables from a copy of the environment
|
|
env = env.copy()
|
|
for key in ["PYTHONPATH", "PYTHONHOME"]:
|
|
env.pop(key, None)
|
|
|
|
return env
|
|
|
|
|
|
def run_openpype_process(*args, **kwargs):
|
|
"""Execute OpenPype process with passed arguments and wait.
|
|
|
|
Wrapper for 'run_process' which prepends OpenPype executable arguments
|
|
before passed arguments and define environments if are not passed.
|
|
|
|
Values from 'os.environ' are used for environments if are not passed.
|
|
They are cleaned using 'clean_envs_for_openpype_process' function.
|
|
|
|
Example:
|
|
```
|
|
run_detached_process("run", "<path to .py script>")
|
|
```
|
|
|
|
Args:
|
|
*args (tuple): OpenPype cli arguments.
|
|
**kwargs (dict): Keyword arguments for for subprocess.Popen.
|
|
"""
|
|
args = get_openpype_execute_args(*args)
|
|
env = kwargs.pop("env", None)
|
|
# Keep env untouched if are passed and not empty
|
|
if not env:
|
|
# Skip envs that can affect OpenPype process
|
|
# - fill more if you find more
|
|
env = clean_envs_for_openpype_process(os.environ)
|
|
|
|
# Only keep OpenPype version if we are running from build.
|
|
if not is_running_from_build():
|
|
env.pop("OPENPYPE_VERSION", None)
|
|
|
|
return run_subprocess(args, env=env, **kwargs)
|
|
|
|
|
|
def run_detached_process(args, **kwargs):
|
|
"""Execute process with passed arguments as separated process.
|
|
|
|
Values from 'os.environ' are used for environments if are not passed.
|
|
They are cleaned using 'clean_envs_for_openpype_process' function.
|
|
|
|
Example:
|
|
```
|
|
run_detached_openpype_process("run", "<path to .py script>")
|
|
```
|
|
|
|
Args:
|
|
*args (tuple): OpenPype cli arguments.
|
|
**kwargs (dict): Keyword arguments for for subprocess.Popen.
|
|
|
|
Returns:
|
|
subprocess.Popen: Pointer to launched process but it is possible that
|
|
launched process is already killed (on linux).
|
|
"""
|
|
env = kwargs.pop("env", None)
|
|
# Keep env untouched if are passed and not empty
|
|
if not env:
|
|
env = os.environ
|
|
|
|
# Create copy of passed env
|
|
kwargs["env"] = {k: v for k, v in env.items()}
|
|
|
|
low_platform = platform.system().lower()
|
|
if low_platform == "darwin":
|
|
new_args = ["open", "-na", args.pop(0), "--args"]
|
|
new_args.extend(args)
|
|
args = new_args
|
|
|
|
elif low_platform == "windows":
|
|
flags = (
|
|
subprocess.CREATE_NEW_PROCESS_GROUP
|
|
| subprocess.DETACHED_PROCESS
|
|
)
|
|
kwargs["creationflags"] = flags
|
|
|
|
if not sys.stdout:
|
|
kwargs["stdout"] = subprocess.DEVNULL
|
|
kwargs["stderr"] = subprocess.DEVNULL
|
|
|
|
elif low_platform == "linux" and get_linux_launcher_args() is not None:
|
|
json_data = {
|
|
"args": args,
|
|
"env": kwargs.pop("env")
|
|
}
|
|
json_temp = tempfile.NamedTemporaryFile(
|
|
mode="w", prefix="op_app_args", suffix=".json", delete=False
|
|
)
|
|
json_temp.close()
|
|
json_temp_filpath = json_temp.name
|
|
with open(json_temp_filpath, "w") as stream:
|
|
json.dump(json_data, stream)
|
|
|
|
new_args = get_linux_launcher_args()
|
|
new_args.append(json_temp_filpath)
|
|
|
|
# Create mid-process which will launch application
|
|
process = subprocess.Popen(new_args, **kwargs)
|
|
# Wait until the process finishes
|
|
# - This is important! The process would stay in "open" state.
|
|
process.wait()
|
|
# Remove the temp file
|
|
os.remove(json_temp_filpath)
|
|
# Return process which is already terminated
|
|
return process
|
|
|
|
process = subprocess.Popen(args, **kwargs)
|
|
return process
|
|
|
|
|
|
def path_to_subprocess_arg(path):
|
|
"""Prepare path for subprocess arguments.
|
|
|
|
Returned path can be wrapped with quotes or kept as is.
|
|
"""
|
|
return subprocess.list2cmdline([path])
|
|
|
|
|
|
def get_pype_execute_args(*args):
|
|
"""Backwards compatible function for 'get_openpype_execute_args'."""
|
|
import traceback
|
|
|
|
log = Logger.get_logger("get_pype_execute_args")
|
|
stack = "\n".join(traceback.format_stack())
|
|
log.warning((
|
|
"Using deprecated function 'get_pype_execute_args'. Called from:\n{}"
|
|
).format(stack))
|
|
return get_openpype_execute_args(*args)
|
|
|
|
|
|
def get_openpype_execute_args(*args):
|
|
"""Arguments to run pype command.
|
|
|
|
Arguments for subprocess when need to spawn new pype process. Which may be
|
|
needed when new python process for pype scripts must be executed in build
|
|
pype.
|
|
|
|
## Why is this needed?
|
|
Pype executed from code has different executable set to virtual env python
|
|
and must have path to script as first argument which is not needed for
|
|
build pype.
|
|
|
|
It is possible to pass any arguments that will be added after pype
|
|
executables.
|
|
"""
|
|
pype_executable = os.environ["OPENPYPE_EXECUTABLE"]
|
|
pype_args = [pype_executable]
|
|
|
|
executable_filename = os.path.basename(pype_executable)
|
|
if "python" in executable_filename.lower():
|
|
pype_args.append(
|
|
os.path.join(os.environ["OPENPYPE_ROOT"], "start.py")
|
|
)
|
|
|
|
if args:
|
|
pype_args.extend(args)
|
|
|
|
return pype_args
|
|
|
|
|
|
def get_linux_launcher_args(*args):
|
|
"""Path to application mid process executable.
|
|
|
|
This function should be able as arguments are different when used
|
|
from code and build.
|
|
|
|
It is possible that this function is used in OpenPype build which does
|
|
not have yet the new executable. In that case 'None' is returned.
|
|
|
|
Args:
|
|
args (iterable): List of additional arguments added after executable
|
|
argument.
|
|
|
|
Returns:
|
|
list: Executables with possible positional argument to script when
|
|
called from code.
|
|
"""
|
|
filename = "app_launcher"
|
|
openpype_executable = os.environ["OPENPYPE_EXECUTABLE"]
|
|
|
|
executable_filename = os.path.basename(openpype_executable)
|
|
if "python" in executable_filename.lower():
|
|
script_path = os.path.join(
|
|
os.environ["OPENPYPE_ROOT"],
|
|
"{}.py".format(filename)
|
|
)
|
|
launch_args = [openpype_executable, script_path]
|
|
else:
|
|
new_executable = os.path.join(
|
|
os.path.dirname(openpype_executable),
|
|
filename
|
|
)
|
|
executable_path = find_executable(new_executable)
|
|
if executable_path is None:
|
|
return None
|
|
launch_args = [executable_path]
|
|
|
|
if args:
|
|
launch_args.extend(args)
|
|
|
|
return launch_args
|