如果我不使用子进程，如何使用 python 读取 stderr？

Question

我正在使用 cwiid 库，这是一个用 C 编写的库，但在 python 中使用。该库允许我使用 Wiimote 来控制机器人上的一些电机。该代码运行作为嵌入式设备上的守护进程，没有显示器、键盘或鼠标。

当我尝试初始化对象时：

import cwiid

while True:
    try:
        wm = cwiid.Wiimote()
    except RuntimeError:
        # RuntimeError exception thrown if no Wiimote is trying to connect

        # Wait a second
        time.sleep(1)

        # Try again
        continue

99% 的时间，一切正常，但偶尔，库会进入某种奇怪的状态，调用 cwiid.Wiimote() 导致库写入 "Socket connect error (control channel)" 到 stderr，python 抛出异常。发生这种情况时，对 cwiid.Wiimote() 的每次后续调用都会导致将相同的内容写入 stderr，并抛出相同的异常，直到我重新启动设备。

我想做的是检测这个问题，然后python自动重启设备。

cwiid库在异常状态下抛出的异常类型也是RuntimeError，这与连接超时异常（这很常见）没有什么不同，所以我似乎无法以这种方式区分它。我想做的是在运行 cwiid.Wiimote() 之后立即读取 stderr 以查看消息 "Socket connect error (control channel)" 是否出现，如果出现，则重新启动。

到目前为止，我可以使用某些 os.dup() 和 os.dup2() 方法重定向 stderr 以防止消息显示，但这似乎无法帮助我阅读 stderr。

大多数在线示例都涉及读取 stderr，如果你是运行带有子进程的东西，这在这种情况下不适用。

我如何着手读取 stderr 以检测写入其中的消息？

我想我要找的是这样的东西：

while True:
    try:
        r, w = os.pipe()
        os.dup2(sys.stderr.fileno(), r)

        wm = cwiid.Wiimote()
    except RuntimeError:
        # RuntimeError exception thrown if no Wiimote is trying to connect

        if ('Socket connect error (control channel)' in os.read(r, 100)):
            # Reboot

        # Wait a second
        time.sleep(1)

        # Try again
        continue

这似乎不像我认为的那样有效。

Answer 1

作为与 stderr 斗争的替代方法，以下方法在放弃之前快速连续重试几次（应该处理连接错误）怎么样：

while True:
    for i in range(50):  # try 50 times
        try:
            wm = cwiid.Wiimote()
            break        # break out of "for" and re-loop in "while"
        except RuntimeError:
            time.sleep(1)
    else:
        raise RuntimeError("permanent Wiimote failure... reboot!")

Answer 2

在幕后，subprocess 除了 dups 之外还使用匿名管道来重定向子进程输出。要让进程读取自己的 stderr，您需要手动执行此操作。它涉及获取一个匿名管道，将标准错误重定向到管道的输入，运行有问题的 stderr 写入操作，从管道的另一端读取输出，并清理所有内容。这一切都很繁琐，但我想我在下面的代码中做对了。

您的 cwiid.Wiimote 调用的以下包装器将 return 一个由函数调用 return 编辑的结果组成的元组（None 在 [=14 的情况下） =]) 和 stderr 输出生成，如果有的话。请参阅 tests 函数，了解它在各种条件下的工作方式。我试着调整您的示例循环，但不太了解 cwiid.Wiimote 调用成功时应该发生什么。在您的示例代码中，您只需立即重新循环。

编辑：糟糕！修复了 example_loop() 中 Wiimote 被调用而不是作为参数传递的错误。

import time

import os
import fcntl

def capture_runtime_stderr(action):
    """Handle runtime errors and capture stderr"""
    (r,w) = os.pipe()
    fcntl.fcntl(w, fcntl.F_SETFL, os.O_NONBLOCK)
    saved_stderr = os.dup(2)
    os.dup2(w, 2)
    try:
        result = action()
    except RuntimeError:
        result = None
    finally:
        os.close(w)
        os.dup2(saved_stderr, 2)
        with os.fdopen(r) as o:
            output = o.read()
    return (result, output)

## some tests

def return_value():
    return 5

def return_value_with_stderr():
    os.system("echo >&2 some output")
    return 10

def runtime_error():
    os.system("echo >&2 runtime error occurred")
    raise RuntimeError()

def tests():
    print(capture_runtime_stderr(return_value))
    print(capture_runtime_stderr(return_value_with_stderr))
    print(capture_runtime_stderr(runtime_error))
    os.system("echo >&2 never fear, stderr is back to normal")

## possible code for your loop

def example_loop():
    while True:
        (wm, output) = capture_runtime_stderr(cmiid.Wiimote)
        if wm == None:
            if "Socket connect error" in output:
                raise RuntimeError("library borked, time to reboot")
            time.sleep(1)
            continue
        ## do something with wm??

如果我不使用子进程，如何使用 python 读取 stderr？

How can I read stderr with python if I'm NOT using subprocess?

python

file-descriptor

stderr

wiimote