Как я могу прочитать stderr с python, если я НЕ использую подпроцесс?

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, который не отличается от исключения тайм-аута соединения (что очень часто встречается), поэтому я не могу отличить его таким образом. То, что я хочу сделать, это прочитать stderr сразу после запуска cwiid.Wiimote(), чтобы узнать, появляется ли сообщение «Ошибка подключения сокета (канал управления)». появляется, и если да, перезагрузитесь.

До сих пор я мог перенаправить stderr, чтобы предотвратить появление сообщения, используя некоторые методы и 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 вызова возвращает кортеж, состоящий из результата, возвращаемого при вызове функции (None в случае RuntimeError) и выход 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

В качестве альтернативы к борьбе с 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!")