且构网

分享程序员开发的那些事...
且构网 - 分享程序员编程开发的那些事

python:基础库日志记录到stdout时使用ncurses

更新时间:2023-12-05 15:01:34

一个最小的演示示例有望显示所有操作方式.我不会仅仅为此设置 SWIG,而是选择通过 ctypes 调用 .so 文件来模拟外部 C 库使用的快速而肮脏的演示.只需将以下内容放入工作目录即可.

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;                                                                   
}

使用 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'))

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 内部,否则我无法找到一种方法来刷新输出以确保全部输出.因此最终的行为可能会很复杂.

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.isAlive():
        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

请注意,所有内容均已捕获.

Note that everything is captured.

现在,由于您确实使用了ncurses ,并且也在线程中运行了该函数,所以这有点棘手.这是龙.

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.

我们将需要ncurses API,该API实际上将使我们能够创建一个新屏幕来重定向输出,并且 ctypes 同样可以很方便地使用.不幸的是,我为系统上的DLL使用了绝对路径.根据需要进行调整.

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

现在我们有了 newterm set_term ,我们终于可以完成脚本了.从主要功能中删除所有内容,然后添加以下内容:

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)

这应该表明在使用ncurses的情况下可以达到预期的结果,但是对于我来说,它最终还是挂了,我不确定还会发生什么.我以为这可能是由于在使用该64位共享库时意外使用32位Python引起的,但是在退出时,某些情况下效果并不理想(我认为滥用 ctypes 很容易,但事实证明确实如此!).无论如何,这至少会像您期望的那样在ncurse窗口中显示输出.

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.