Executable Called Via Subprocess.check_output Prints On Console But Result Is Not Returned
On a Windows machine, I'm trying to call an external executable from Python and gather its outputs for further processing. Because a local path variable has to be set before callin
Solution 1:
If a program writes directly to the console (e.g. by opening the CONOUT$ device) instead of to the process standard handles, the only option is to read the console screen buffer directly. To make this simpler, start with a new, empty screen buffer. Create, size, initialize, and activate a new screen buffer via the following functions:
CreateConsoleScreenBufferGetConsoleScreenBufferInfoEx(Minimum supported client: Windows Vista)SetConsoleScreenBufferInfoEx(Minimum supported client: Windows Vista)SetConsoleWindowInfoFillConsoleOutputCharacterSetConsoleActiveScreenBuffer
Make sure to request GENERIC_READ | GENERIC_WRITE access when calling CreateConsoleScreenBuffer. You'll need read access later in order to read the contents of the screen.
Specifically for Python, use ctypes to call functions in the Windows console API. Also, if you wrap the handle with a C file descriptor via msvcrt.open_osfhandle, then you can pass it as the stdout or stderr argument of subprocess.Popen.
The file descriptor or handle for the screen buffer can't be read directly via read, ReadFile, or even ReadConsole. If you have a file descriptor, get the underlying handle via msvcrt.get_osfhandle. Given a screen buffer handle, call ReadConsoleOutputCharacter to read from the screen. The read_screen function in the sample code below demonstrates reading from the beginning of the screen buffer up to the cursor position.
A process needs to be attached to a console in order to use the console API. To that end, I've included a simple allocate_console context manager to temporarily open a console. This is useful in a GUI application, which normally isn't attached to a console.
The following example was tested in Windows 7 and 10, in Python 2.7 and 3.5.
ctypes definitions
import os
import contextlib
import msvcrt
import ctypes
from ctypes import wintypes
kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)
GENERIC_READ = 0x80000000
GENERIC_WRITE = 0x40000000
FILE_SHARE_READ = 1
FILE_SHARE_WRITE = 2
CONSOLE_TEXTMODE_BUFFER = 1
INVALID_HANDLE_VALUE = wintypes.HANDLE(-1).value
STD_OUTPUT_HANDLE = wintypes.DWORD(-11)
STD_ERROR_HANDLE = wintypes.DWORD(-12)
def_check_zero(result, func, args):
ifnot result:
raise ctypes.WinError(ctypes.get_last_error())
return args
def_check_invalid(result, func, args):
if result == INVALID_HANDLE_VALUE:
raise ctypes.WinError(ctypes.get_last_error())
return args
ifnothasattr(wintypes, 'LPDWORD'): # Python 2
wintypes.LPDWORD = ctypes.POINTER(wintypes.DWORD)
wintypes.PSMALL_RECT = ctypes.POINTER(wintypes.SMALL_RECT)
classCOORD(ctypes.Structure):
_fields_ = (('X', wintypes.SHORT),
('Y', wintypes.SHORT))
classCONSOLE_SCREEN_BUFFER_INFOEX(ctypes.Structure):
_fields_ = (('cbSize', wintypes.ULONG),
('dwSize', COORD),
('dwCursorPosition', COORD),
('wAttributes', wintypes.WORD),
('srWindow', wintypes.SMALL_RECT),
('dwMaximumWindowSize', COORD),
('wPopupAttributes', wintypes.WORD),
('bFullscreenSupported', wintypes.BOOL),
('ColorTable', wintypes.DWORD * 16))
def__init__(self, *args, **kwds):
super(CONSOLE_SCREEN_BUFFER_INFOEX, self).__init__(
*args, **kwds)
self.cbSize = ctypes.sizeof(self)
PCONSOLE_SCREEN_BUFFER_INFOEX = ctypes.POINTER(
CONSOLE_SCREEN_BUFFER_INFOEX)
LPSECURITY_ATTRIBUTES = wintypes.LPVOID
kernel32.GetStdHandle.errcheck = _check_invalid
kernel32.GetStdHandle.restype = wintypes.HANDLE
kernel32.GetStdHandle.argtypes = (
wintypes.DWORD,) # _In_ nStdHandle
kernel32.CreateConsoleScreenBuffer.errcheck = _check_invalid
kernel32.CreateConsoleScreenBuffer.restype = wintypes.HANDLE
kernel32.CreateConsoleScreenBuffer.argtypes = (
wintypes.DWORD, # _In_ dwDesiredAccess
wintypes.DWORD, # _In_ dwShareMode
LPSECURITY_ATTRIBUTES, # _In_opt_ lpSecurityAttributes
wintypes.DWORD, # _In_ dwFlags
wintypes.LPVOID) # _Reserved_ lpScreenBufferData
kernel32.GetConsoleScreenBufferInfoEx.errcheck = _check_zero
kernel32.GetConsoleScreenBufferInfoEx.argtypes = (
wintypes.HANDLE, # _In_ hConsoleOutput
PCONSOLE_SCREEN_BUFFER_INFOEX) # _Out_ lpConsoleScreenBufferInfo
kernel32.SetConsoleScreenBufferInfoEx.errcheck = _check_zero
kernel32.SetConsoleScreenBufferInfoEx.argtypes = (
wintypes.HANDLE, # _In_ hConsoleOutput
PCONSOLE_SCREEN_BUFFER_INFOEX) # _In_ lpConsoleScreenBufferInfo
kernel32.SetConsoleWindowInfo.errcheck = _check_zero
kernel32.SetConsoleWindowInfo.argtypes = (
wintypes.HANDLE, # _In_ hConsoleOutput
wintypes.BOOL, # _In_ bAbsolute
wintypes.PSMALL_RECT) # _In_ lpConsoleWindow
kernel32.FillConsoleOutputCharacterW.errcheck = _check_zero
kernel32.FillConsoleOutputCharacterW.argtypes = (
wintypes.HANDLE, # _In_ hConsoleOutput
wintypes.WCHAR, # _In_ cCharacter
wintypes.DWORD, # _In_ nLength
COORD, # _In_ dwWriteCoord
wintypes.LPDWORD) # _Out_ lpNumberOfCharsWritten
kernel32.ReadConsoleOutputCharacterW.errcheck = _check_zero
kernel32.ReadConsoleOutputCharacterW.argtypes = (
wintypes.HANDLE, # _In_ hConsoleOutput
wintypes.LPWSTR, # _Out_ lpCharacter
wintypes.DWORD, # _In_ nLength
COORD, # _In_ dwReadCoord
wintypes.LPDWORD) # _Out_ lpNumberOfCharsReadfunctions
@contextlib.contextmanagerdefallocate_console():
allocated = kernel32.AllocConsole()
try:
yield allocated
finally:
if allocated:
kernel32.FreeConsole()
@contextlib.contextmanagerdefconsole_screen(ncols=None, nrows=None):
info = CONSOLE_SCREEN_BUFFER_INFOEX()
new_info = CONSOLE_SCREEN_BUFFER_INFOEX()
nwritten = (wintypes.DWORD * 1)()
hStdOut = kernel32.GetStdHandle(STD_OUTPUT_HANDLE)
kernel32.GetConsoleScreenBufferInfoEx(
hStdOut, ctypes.byref(info))
if ncols isNone:
ncols = info.dwSize.X
if nrows isNone:
nrows = info.dwSize.Y
elif nrows > 9999:
raise ValueError('nrows must be 9999 or less')
fd_screen = None
hScreen = kernel32.CreateConsoleScreenBuffer(
GENERIC_READ | GENERIC_WRITE,
FILE_SHARE_READ | FILE_SHARE_WRITE,
None, CONSOLE_TEXTMODE_BUFFER, None)
try:
fd_screen = msvcrt.open_osfhandle(
hScreen, os.O_RDWR | os.O_BINARY)
kernel32.GetConsoleScreenBufferInfoEx(
hScreen, ctypes.byref(new_info))
new_info.dwSize = COORD(ncols, nrows)
new_info.srWindow = wintypes.SMALL_RECT(
Left=0, Top=0, Right=(ncols - 1),
Bottom=(info.srWindow.Bottom - info.srWindow.Top))
kernel32.SetConsoleScreenBufferInfoEx(
hScreen, ctypes.byref(new_info))
kernel32.SetConsoleWindowInfo(hScreen, True,
ctypes.byref(new_info.srWindow))
kernel32.FillConsoleOutputCharacterW(
hScreen, u'\0', ncols * nrows, COORD(0,0), nwritten)
kernel32.SetConsoleActiveScreenBuffer(hScreen)
try:
yield fd_screen
finally:
kernel32.SetConsoleScreenBufferInfoEx(
hStdOut, ctypes.byref(info))
kernel32.SetConsoleWindowInfo(hStdOut, True,
ctypes.byref(info.srWindow))
kernel32.SetConsoleActiveScreenBuffer(hStdOut)
finally:
if fd_screen isnotNone:
os.close(fd_screen)
else:
kernel32.CloseHandle(hScreen)
defread_screen(fd):
hScreen = msvcrt.get_osfhandle(fd)
csbi = CONSOLE_SCREEN_BUFFER_INFOEX()
kernel32.GetConsoleScreenBufferInfoEx(
hScreen, ctypes.byref(csbi))
ncols = csbi.dwSize.X
pos = csbi.dwCursorPosition
length = ncols * pos.Y + pos.X + 1
buf = (ctypes.c_wchar * length)()
n = (wintypes.DWORD * 1)()
kernel32.ReadConsoleOutputCharacterW(
hScreen, buf, length, COORD(0,0), n)
lines = [buf[i:i+ncols].rstrip(u'\0')
for i inrange(0, n[0], ncols)]
returnu'\n'.join(lines)
example
if __name__ == '__main__':
import io
import textwrap
import subprocess
text = textwrap.dedent('''\
Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do
eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut
enim ad minim veniam, quis nostrud exercitation ullamco laboris
nisi ut aliquip ex ea commodo consequat. Duis aute irure dolor
in reprehenderit in voluptate velit esse cillum dolore eu
fugiat nulla pariatur. Excepteur sint occaecat cupidatat non
proident, sunt in culpa qui officia deserunt mollit anim id est
laborum.''')
cmd = ("python -c \"""print('piped output');""conout = open(r'CONOUT$', 'w');""conout.write('''%s''')\"" % text)
with allocate_console() as allocated:
with console_screen(nrows=1000) as fd_conout:
stdout = subprocess.check_output(cmd).decode()
conout = read_screen(fd_conout)
with io.open('result.txt', 'w', encoding='utf-8') as f:
f.write(u'stdout:\n' + stdout)
f.write(u'\nconout:\n' + conout)
output
stdout:
piped output
conout:
Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do
eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut
enim ad minim veniam, quis nostrud exercitation ullamco laboris
nisi ut aliquip ex ea commodo consequat. Duis aute irure dolor
in reprehenderit in voluptate velit esse cillum dolore eu
fugiat nulla pariatur. Excepteur sint occaecat cupidatat non
proident, sunt in culpa qui officia deserunt mollit anim id est
laborum.
Post a Comment for "Executable Called Via Subprocess.check_output Prints On Console But Result Is Not Returned"