如何使用curses在终端中编写三个不断更新的行？

Question

我有一个程序，需要输出三个不断变化的消息。连接数、经过时间和被拒绝的连接数。

我尝试在字符串末尾使用`'\r'`并在其他消息开始它们自己的输出循环之前打印换行符，以为回车会回到上一行。但是它们最终都在第一行上互相覆盖了。

我看到类似的问题，人们推荐使用`curses`，但是我似乎无法让它工作。我尝试使用`addstr()`而不是`print`或`sys.stdout.write()`，但显然我做错了，因为它最终看起来像this.。我还尝试在每个`addstr`之后使用`move(0,0)`，因为我认为坐标是从光标的最后位置计算的，但输出看起来与我没有这样做时相同。

以下是我当前使用`curses`的代码 -

from scapy.all import *
from curses import wrapper 
import argparse, random, socket, sys, time, os, threading, re, subprocess, curses

def main(target):
    try: 
        stdscr.clear()
        curses.noecho()
        curses.cbreak()
        stdscr.keypad(1)

        #things regarding user agents and iptables

            for conn in range(args.connections):
                t = threading.Thread(target=sendPacket, args=(targetIP, args.port, targetHost,random.choice(userAgents)))
                threads.append(t)
                t.start()
                numConn += 1

                try:
                    lock.acquire()
                    stdscr.addstr(0, 0,'{0} Connections Established to {1} '.format(numConn,getIP(target)))
                    stdscr.refresh()
                    #sys.stdout.flush()
                finally:
                    lock.release()
            time.sleep(args.timer)  


        CloseWin()
        sys.exit()
    except KeyboardInterrupt, e:
        lock.acquire()
        stdscr.addstr (5,0,'Exiting {0}'.format(e))
        CloseWin()
        lock.release()
        sys.exit()

#functions that deal with resolving host and IP

def sendPacket(targetIP,targetPort,targetHost,userAgent):
    try:
        #building and sending the packet
        failPacket = 0
        lock.acquire()
        if synack is None:
            stdscr.addstr(1, 0, '{0} Connections refused'.format(failPacket))
            stdscr.refresh()
            failPacket += 1
            return

        #send final packet

        lock.release()
        return
    except Exception,e:
        CloseWin()
        #print e
        lock.release()
        return 



def MeasureTime():
try:
    lock.acquire()
    while True:
        stdscr.addstr(3, 0,'{0} Time elapsed'.format(time.time() - startTime))
        stdscr.refresh()
finally:
    lock.release()

def CloseWin():
try:
    lock.acquire()
    curses.nocbreak()
    stdscr.keypad(0)
    curses.echo()
    curses.endwin()
finally:
    lock.release()

if __name__ == '__main__':

    #args

    stdscr = curses.initscr()

    if args.debug == True:
        wrapper(OneConn(args.target)) #OneConn is the same as the main function except it only opens one connection, so the print is the same
    else:
        startTime = time.time()
        wrapper(main(args.target))

此外，除了不能正确编写外，我似乎无法正确关闭它。当我使用ctrl+c尝试退出时，看起来它停止更新有关数据包的打印输出，但时间仍在继续。我找不到其他停止它的方法，除了关闭终端。

根据@martinaeu的建议尝试添加锁，并编辑了帖子以显示我所做的内容。结果是，我只能看到第四行中打印出并更新的经过的时间，完全看不到其他打印内容，并且仍然无法退出程序而不关闭终端。

- GoodKingRene

---

你遇到的问题可能与多线程有关，因为有多个线程同时（逻辑上至少）尝试使用curses库。要成功并发地使用它，你可能需要添加一个Lock对象，在每个线程（包括主线程）在使用curses API之前必须获取该锁，并在使用完毕后释放它。 - martineau

你使用Lock()的方式有点低效，并且在多线程的sendPacket()函数中可能存在一个小bug。建议你多读一些关于它们的资料，并开始使用上下文管理器（通过with语句）。这样可以消除大部分（如果不是全部）try/finally的代码（因为with会确保无论是否有异常，都会调用release()，即使有多个return语句）。 - martineau

关于原始问题，我认为addstr()的位置参数是相对于它们发送输出到的屏幕角落的，这意味着每个sendPacket线程都需要在一个不同的行上添加/更新信息，以便它们都可见。当创建每个线程时，您可以将行（在这里并不真正需要列）位置作为额外的参数传递给函数。 - martineau

@martineau，我想让 sendPacket 在一行中写出所有内容，只需更改发送的数据包数量。一开始我以为 addstr() 是相对于屏幕角落的，而且我认为我调用行数的方式是有意义的，但似乎并不起作用。关于 lock()，我是否只需用 with lock: 替换所有的 try/finally 代码块？因为我尝试过这样做，我很确定自己没有出错，但并没有起作用。 - GoodKingRene

我觉得我明白你想要的是，即更新已经显示在多行/行上的内容。为了做到这一点，每个线程都需要被“分配”一行/行，并且它应该始终在调用addstr()时使用该行。 - martineau

Answer 1

不必使用curses，您可以直接发出原始代码将光标向后移动三行。代码如下：

os.write(1, "\x1b[3F")

注意：这些行并没有被擦除。如果你打印的新行与旧行具有相同的长度（例如，因为它们是用类似于"foo：%10d"的格式进行格式化），那么这不应该是一个问题。

参考资料：https://en.wikipedia.org/wiki/ANSI_escape_code#CSI_codes（其中“CSI”代表“\x1b [”，n是以十进制写成的可选数字）。

- Armin Rigo

---

谢谢你的回答，我之前不知道这是一个功能。对于python3，需要一个字节对象，例如os.write(1, b'\x1b[3F')。 - kat

“3F”中的“3”表示您想要回退的行数，例如，“4F”表示回退4行。 - Richard