subprocess_runner.py 2.5 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576
  1. #
  2. # Copyright (c) Contributors to the Open 3D Engine Project.
  3. # For complete copyright and license terms please see the LICENSE at the root of this distribution.
  4. #
  5. # SPDX-License-Identifier: Apache-2.0 OR MIT
  6. #
  7. import subprocess
  8. class SubprocessRunner:
  9. """
  10. Class for common methods and properties
  11. required to dispatch a shell/system/python application
  12. and capture the errorCode, stdout and stderr
  13. """
  14. def __init__(self, argList: list[str], timeOutSeconds: int, cwd: str = ".") :
  15. self._name = self.__class__.__name__
  16. self._argList = argList
  17. self._arg_list_str = "" # Becomes the command string when executed.
  18. self._timeOut = timeOutSeconds
  19. self._error_code = -1
  20. self._error_message = ""
  21. self._success_message = ""
  22. self._cwd = cwd
  23. def run(self) -> bool:
  24. """
  25. This is a blocking call that returns when the subprocess is finished.
  26. @returns True if the system error code is 0 (success).
  27. """
  28. self._arg_list_str = " ".join(self._argList)
  29. print(f"{self._name} will run command:\n{self._arg_list_str}\n")
  30. self._subprocess = subprocess.Popen(
  31. self._argList, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=self._cwd
  32. )
  33. try:
  34. outs, errs = self._subprocess.communicate(timeout=self._timeOut)
  35. self._success_message = outs.decode("utf-8")
  36. self._error_message = errs.decode("utf-8")
  37. print(f"ok:<{self._success_message}>, err:<{self._error_message}>")
  38. self._error_code = self._subprocess.returncode
  39. return self._subprocess.returncode == 0
  40. except subprocess.TimeoutExpired:
  41. self._subprocess.kill()
  42. outs, errs = self._subprocess.communicate()
  43. self._success_message = outs.decode("utf-8")
  44. self._error_message = errs.decode("utf-8")
  45. print(f"ok:<{self._success_message}>, err:<{self._error_message}>")
  46. self._error_code = -1
  47. return False
  48. def get_error_code(self):
  49. return self._error_code
  50. def get_stderr(self) -> str:
  51. return self._error_message
  52. def get_stdout(self) -> str:
  53. return self._success_message
  54. def get_stdall(self) -> str:
  55. return f"{self._success_message}\n{self._error_message}"
  56. def get_command_str(self) -> str:
  57. return self._arg_list_str
  58. # class SubprocessRunner END
  59. ######################################################