I am trying to write a small python program that uses curses and a SWIGed C++ library. That library logs a lot of information to STDOUT, which interferes with the output from curses. I would like to somehow intercept that content and then display it nicely through ncurses. Is there some way to do this?
A minimal demonstrating example will hopefully show how this all works. I am not going to set up SWIG just for this, and opt for a quick and dirty demonstration of calling a .so file through ctypes to emulate that external C library usage. Just put the following in the working directory.
#include <stdio.h>
int vomit(void);
int vomit()
{
printf("vomiting output onto stdout\n");
fflush(stdout);
return 1;
}
Build with gcc -shared -Wl,-soname,testlib -o _testlib.so -fPIC testlib.c
import ctypes
from os.path import dirname
from os.path import join
testlib = ctypes.CDLL(join(dirname(__file__), '_testlib.so'))
import os
import sys
import testlib
from tempfile import mktemp
pipename = mktemp()
os.mkfifo(pipename)
pipe_fno = os.open(pipename, os.O_RDWR | os.O_NONBLOCK)
stdout_fno = os.dup(sys.stdout.fileno())
os.dup2(pipe_fno, 1)
result = testlib.testlib.vomit()
os.dup2(stdout_fno, 1)
buf = bytearray()
while True:
try:
buf += os.read(pipe_fno, 1)
except Exception:
break
print("the captured output is: %s" % open('scratch').read())
print('the result of the program is: %d' % result)
os.unlink(pipename)
The caveat is that the output generated by the .so might be buffered somehow within the ctypes system (I have no idea how that part all works), and I cannot find a way to flush the output to ensure they are all outputted unless the fflush code is inside the .so; so there can be complications with how this ultimately behaves.
With threading, this can be done also (code is becoming quite atrocious, but it shows the idea):
import os
import sys
import testlib
from threading import Thread
from time import sleep
from tempfile import mktemp
def external():
# the thread that will call the .so that produces output
for i in range(7):
testlib.testlib.vomit()
sleep(1)
# setup
stdout_fno = os.dup(sys.stdout.fileno())
pipename = mktemp()
os.mkfifo(pipename)
pipe_fno = os.open(pipename, os.O_RDWR | os.O_NONBLOCK)
os.dup2(pipe_fno, 1)
def main():
thread = Thread(target=external)
thread.start()
buf = bytearray()
counter = 0
while thread.is_alive():
sleep(0.2)
try:
while True:
buf += os.read(pipe_fno, 1)
except BlockingIOError:
if buf:
# do some processing to show that the string is fully
# captured
output = 'external lib: [%s]\n' % buf.strip().decode('utf8')
# low level write to original stdout
os.write(stdout_fno, output.encode('utf8'))
buf.clear()
os.write(stdout_fno, b'tick: %d\n' % counter)
counter += 1
main()
# cleanup
os.dup2(stdout_fno, 1)
os.close(pipe_fno)
os.unlink(pipename)
Example execution:
$ python demo2.py
external lib: [vomiting output onto stdout]
tick: 0
tick: 1
tick: 2
tick: 3
external lib: [vomiting output onto stdout]
tick: 4
Note that everything is captured.
Now, since you do have make use of ncurses and also run that function in a thread, this is a bit tricky. Here be dragons.
We will need the ncurses API that will actually let us create a new screen to redirect the output, and again ctypes can be handy for this. Unfortunately, I am using absolute paths for the DLLs on my system; adjust as required.
import ctypes
libc = ctypes.CDLL('/lib64/libc.so.6')
ncurses = ctypes.CDLL('/lib64/libncursesw.so.6')
class FILE(ctypes.Structure):
pass
class SCREEN(ctypes.Structure):
pass
FILE_p = ctypes.POINTER(FILE)
libc.fdopen.restype = FILE_p
SCREEN_p = ctypes.POINTER(SCREEN)
ncurses.newterm.restype = SCREEN_p
ncurses.set_term.restype = SCREEN_p
fdopen = libc.fdopen
newterm = ncurses.newterm
set_term = ncurses.set_term
delscreen = ncurses.delscreen
endwin = ncurses.endwin
Now that we have newterm and set_term, we can finally complete the script. Remove everything from the main function, and add the following:
# setup the curse window
import curses
from lib import newterm, fdopen, set_term, endwin, delscreen
stdin_fno = sys.stdin.fileno()
stdscr = curses.initscr()
# use the ctypes library to create a new screen and redirect output
# back to the original stdout
screen = newterm(None, fdopen(stdout_fno, 'w'), fdopen(stdin_fno, 'r'))
old_screen = set_term(screen)
stdscr.clear()
curses.noecho()
border = curses.newwin(8, 68, 4, 4)
border.border()
window = curses.newwin(6, 66, 5, 5)
window.scrollok(True)
window.clear()
border.refresh()
window.refresh()
def main():
thread = Thread(target=external)
thread.start()
buf = bytearray()
counter = 0
while thread.isAlive():
sleep(0.2)
try:
while True:
buf += os.read(pipe_fno, 1)
except BlockingIOError:
if buf:
output = 'external lib: [%s]\n' % buf.strip().decode('utf8')
buf.clear()
window.addstr(output)
window.refresh()
window.addstr('tick: %d\n' % counter)
counter += 1
window.refresh()
main()
# cleanup
os.dup2(stdout_fno, 1)
endwin()
delscreen(screen)
os.close(pipe_fno)
os.unlink(pipename)
This should sort of show that the intended result with the usage of ncurses be achieved, however for my case it hung at the end and I am not sure what else might be going on. I thought this could be caused by an accidental use of 32-bit Python while using that 64-bit shared object, but on exit things somehow don't play nicely (I thought misuse of ctypes is easy, but turns out it really is!). Anyway, this least it shows the output inside an ncurse window as you might expect.