python: Using ncurses when underlying library logs to stdout

Question

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?

Answer 1

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.

testlib.c

#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

testlib.py

import ctypes                                                                   
from os.path import dirname                                                     
from os.path import join                                                        
                                                                                
testlib = ctypes.CDLL(join(dirname(__file__), '_testlib.so'))

demo.py (for minimum demonstration)

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.

lib.py

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.

Answer 2

@metatoaster indicated a link which talks about a way to temporarily redirect the standard output to /dev/null. That could show something about how to use dup2, but is not quite an answer by itself.

python's interface to curses uses only initscr, which means that the curses library writes its output to the standard output. The SWIG'd library writes its output to the standard output, but that would interfere with the curses output. You could solve the problem by

• redirecting the curses output to /dev/tty, and
• redirecting the SWIG'd output to a temporary file, and
• reading the file, checking for updates to add to the screen.

Once initscr has been called, the curses library has its own copy of the output stream. If you can temporarily point the real standard output to a file first (before initializing curses), then open a new standard output to /dev/tty (for initscr), and then restore the (global!) output stream then that should work.