python: 当底层库记录到标准输出时使用 ncurses

Question

我正在尝试编写一个使用 curses 和 SWIGed C++ 库的小 python 程序。该库将大量信息记录到 STDOUT，这会干扰 curses 的输出。我想以某种方式拦截该内容，然后通过 ncurses 很好地显示它。有什么办法吗？

Answer 1

@metatoaster 表示 link 讨论了一种将标准输出临时重定向到 /dev/null 的方法。这可能会显示一些关于如何使用 dup2 的信息，但它本身并不是一个完整的答案。

python 与 curses 的接口仅使用 initscr，这意味着 curses 库将其输出写入标准输出。 SWIG 库将其输出写入标准输出，但这会干扰 curses 输出。您可以通过

解决问题

将 curses 输出重定向到 /dev/tty，并且
将 SWIG 的输出重定向到一个临时文件，并且
正在读取文件，检查要添加到屏幕的更新。

调用 initscr 后，curses 库就有了自己的输出流副本。如果您可以先将真正的标准输出暂时指向一个文件（在初始化 curses 之前），然后打开一个新的标准输出到 /dev/tty（对于 initscr），然后恢复（全局！）输出流那应该可以了。

Answer 2

一个最小的演示示例有望展示这一切是如何工作的。我不打算为此设置 SWIG，而是选择通过 ctypes 调用 .so 文件来模拟外部 C 库用法的快速而肮脏的演示。只需将以下内容放入工作目录即可。

testlib.c

#include <stdio.h>

int vomit(void);                                                                
                                                                                
int vomit()                                                                     
{                                                                               
    printf("vomiting output onto stdout\n");                                    
    fflush(stdout);                                                             
    return 1;                                                                   
}

使用 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（用于最小演示）

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)

需要注意的是 .so 生成的输出可能 在 ctypes 系统中以某种方式被缓冲（我不知道这部分是怎么回事有效），我找不到刷新输出以确保它们全部输出的方法，除非 fflush 代码在 .so; 内。所以这最终的行为可能会很复杂。

使用线程，这也可以完成（代码变得非常糟糕，但它显示了这个想法）：

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)

执行示例：

$ 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

请注意，所有内容都已捕获。

现在，由于您已经在线程中使用了 ncurses 和运行，这有点棘手。这里有龙。

我们将需要 ncurses API，它实际上可以让我们创建一个新屏幕来重定向输出，ctypes 也可以派上用场。不幸的是，我在系统上使用 DLL 的绝对路径；根据需要调整。

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

现在我们有了 newterm 和 set_term，我们终于可以完成脚本了。从 main 函数中删除所有内容，并添加以下内容：

# 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)

这应该表明使用 ncurses 可以达到预期的结果，但是对于我的情况，它在最后挂起，我不确定还会发生什么。我认为这可能是由于在使用 64 位共享对象时意外使用了 32 位 Python 引起的，但是在退出时不知何故不能很好地播放（我认为 ctypes 的误用很容易，但事实证明确实如此！）。无论如何，这至少如您所料，它显示了 ncurse window 中的输出。

python: 当底层库记录到标准输出时使用 ncurses

python: Using ncurses when underlying library logs to stdout

python

curses

logging

stdout

testlib.c

testlib.py

demo.py（用于最小演示）

lib.py