如果我不使用子進程，如何用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()結果庫編寫「套接字連接錯誤（控制通道）」到stderr，並且python拋出異常。發生這種情況時，每次後續調用cwiid.Wiimote()都會導致同樣的事情被寫入stderr，並且在重新啓動設備之前拋出相同的異常。

我想要做的是檢測到這個問題，並讓python自動重啓設備。

cwiid庫在怪異狀態下拋出異常的類型也是RuntimeError，這與連接超時異常（這很常見）沒有什麼不同，所以我似乎無法區分它。我想要做的就是在運行cwiid.Wiimote()之後立即閱讀stderr，看看是否顯示消息「套接字連接錯誤（控制通道）」出現，如果有，請重新啓動。

到目前爲止，我可以重定向stderr以防止消息通過使用某些os.dup()和os.dup2()方法顯示出來，但這看起來不會幫助我閱讀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

引擎蓋下，subprocess使用匿名管道除了重新定向子流程輸出。爲了讓程序讀取它自己的stderr，你需要手動完成。它涉及到獲取匿名管道，將標準錯誤重定向到管道輸入，運行有問題的stderr寫入操作，讀取管道另一端的輸出以及清除所有內容。這一切都相當繁瑣，但我認爲我在下面的代碼中正確無誤。

cwiid.Wiimote調用的以下包裝將返回由函數調用返回的結果（如果是RuntimeError，則返回None）和生成的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?? 

Answer 2

作爲替代與標準錯誤的戰鬥，怎麼樣它放棄之前重試幾次快速連續地（應該處理連接錯誤）如下：

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!")