Last active
June 25, 2022 11:00
-
-
Save SCP002/6c21348635264a90c0cc7b642c446399 to your computer and use it in GitHub Desktop.
Python: Start process. Keep StdOut and StdErr in the original order. Display output real time character by character. Capture output on exit.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/python3 | |
# -*- coding: utf-8 -*- | |
# Requirements: | |
# Python 3.7+ (uses data classes, lines 16 and 25). | |
import dataclasses as dc | |
import os | |
import subprocess | |
import sys | |
import traceback | |
from collections.abc import Sequence | |
from typing import TextIO, Union | |
@dc.dataclass() | |
class ProcessOptions: | |
command: Sequence[str] | |
wait: bool = True | |
shell: bool = False | |
new_console: bool = False | |
hide: bool = False | |
@dc.dataclass() | |
class ProcessResult: | |
exit_code: int = -1 | |
output: str = '' | |
class App: | |
@staticmethod | |
def start_process(opts: ProcessOptions) -> ProcessResult: | |
result: ProcessResult = ProcessResult() | |
startup_info: subprocess.STARTUPINFO = subprocess.STARTUPINFO() | |
creation_flags: int = 0 | |
std_out: Union[int, TextIO, None] = None | |
std_err: Union[int, TextIO, None] = None | |
std_in: Union[int, TextIO, None] = None | |
if opts.new_console or opts.hide: | |
if opts.new_console: | |
creation_flags |= subprocess.CREATE_NEW_CONSOLE | |
if opts.hide: | |
startup_info.dwFlags |= subprocess.STARTF_USESHOWWINDOW | |
else: # Can capture output... | |
std_out = subprocess.PIPE | |
std_err = subprocess.STDOUT | |
# Fix "ERROR: Input redirection is not supported, exiting the process immediately" on Windows: | |
std_in = sys.stdin | |
proc: subprocess.Popen[str] = subprocess.Popen( | |
opts.command, | |
encoding='utf-8', | |
shell=opts.shell, | |
startupinfo=startup_info, | |
creationflags=creation_flags, | |
stdout=std_out, | |
stderr=std_err, | |
stdin=std_in, | |
) | |
can_capture_out: bool = not opts.new_console and not opts.hide | |
if can_capture_out and opts.wait: | |
while True: | |
char = proc.stdout.read(1) | |
if char == '' and proc.poll() is not None: | |
break | |
if char: | |
result.output += char | |
sys.stdout.write(char) | |
sys.stdout.flush() | |
if opts.wait: | |
proc.wait() | |
result.exit_code = proc.poll() | |
return result | |
@staticmethod | |
def main() -> None: | |
opts: ProcessOptions = ProcessOptions( | |
command=('my-executable-name', 'arg1'), | |
wait=True, | |
# shell=False, | |
# new_console=False, | |
# hide=False, | |
) | |
result: ProcessResult = App.start_process(opts) | |
print('-' * 30) | |
print(result.output) | |
print('Exit code:', result.exit_code) | |
input('Press <Enter> to exit...\n') | |
# Main start point. | |
if __name__ == '__main__': | |
# noinspection PyBroadException | |
try: | |
os.chdir(sys.path[0]) | |
App.main() | |
except Exception: | |
traceback.print_exc() | |
input('Press <Enter> to exit...\n') | |
exit(1) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment