Nicht blockierendes Lesen eines Unterprozesses. PIPE in Python

506

Ich verwende das Unterprozessmodul , um einen Unterprozess zu starten und eine Verbindung zu seinem Ausgabestream (stdout) herzustellen. Ich möchte in der Lage sein, nicht blockierende Lesevorgänge auf seinem Standard auszuführen. Gibt es eine Möglichkeit, .readline nicht zu blockieren oder zu überprüfen, ob sich Daten im Stream befinden, bevor ich sie aufrufe .readline? Ich möchte, dass dies portabel ist oder zumindest unter Windows und Linux funktioniert.

Hier ist, wie ich es jetzt mache (es blockiert, .readlinewenn keine Daten verfügbar sind):

p = subprocess.Popen('myprogram.exe', stdout = subprocess.PIPE)
output_str = p.stdout.readline()
Mathieu Pagé
quelle
14
(Von Google?) Alle PIPEs blockieren, wenn einer der PIPEs-Puffer voll ist und nicht gelesen wird. zB stdout Deadlock, wenn stderr gefüllt ist. Übergeben Sie niemals ein Rohr, das Sie nicht lesen möchten.
Nasser Al-Wohaibi
@ NasserAl-Wohaibi bedeutet das, dass es besser ist, immer Dateien zu erstellen?
Charlie Parker
Ich war neugierig zu verstehen, warum es überhaupt blockiert ... Ich frage, weil ich den Kommentar gesehen habe:To avoid deadlocks: careful to: add \n to output, flush output, use readline() rather than read()
Charlie Parker
Es wartet "von Natur aus" darauf, Eingaben zu erhalten.
Mathieu Pagé
Verwandte: stackoverflow.com/q/19880190/240515
user240515

Antworten:

403

fcntl, select, asyncprocWird in diesem Fall nicht helfen.

Eine zuverlässige Methode zum Lesen eines Streams ohne Blockierung unabhängig vom Betriebssystem ist die Verwendung von Queue.get_nowait():

import sys
from subprocess import PIPE, Popen
from threading  import Thread

try:
    from queue import Queue, Empty
except ImportError:
    from Queue import Queue, Empty  # python 2.x

ON_POSIX = 'posix' in sys.builtin_module_names

def enqueue_output(out, queue):
    for line in iter(out.readline, b''):
        queue.put(line)
    out.close()

p = Popen(['myprogram.exe'], stdout=PIPE, bufsize=1, close_fds=ON_POSIX)
q = Queue()
t = Thread(target=enqueue_output, args=(p.stdout, q))
t.daemon = True # thread dies with the program
t.start()

# ... do other things here

# read line without blocking
try:  line = q.get_nowait() # or q.get(timeout=.1)
except Empty:
    print('no output yet')
else: # got line
    # ... do something with line
jfs
quelle
6
Ja, das funktioniert bei mir, aber ich habe viel entfernt. Es enthält bewährte Verfahren, die jedoch nicht immer erforderlich sind. Python 3.x 2.X-kompatibel und close_fds können weggelassen werden, es funktioniert weiterhin. Aber sei dir nur bewusst, was alles macht und kopiere es nicht blind, auch wenn es einfach funktioniert! (Eigentlich ist die einfachste Lösung, einen Thread zu verwenden und eine Readline zu machen, wie es Seb getan hat. Fragen sind nur ein einfacher Weg, um die Daten zu erhalten, es gibt andere, Threads sind die Antwort!)
Aki
3
Innerhalb des Threads out.readlineblockiert der Aufruf, den Thread und den Hauptthread zu blockieren, und ich muss warten, bis readline zurückkehrt, bevor alles andere fortgesetzt wird. Irgendein einfacher Weg, das zu umgehen? (Ich lese mehrere Zeilen aus meinem Prozess, der auch eine andere .py-Datei ist, die DB und Dinge tut)
Justin
3
@Justin: 'out.readline' blockiert nicht den Hauptthread, der in einem anderen Thread ausgeführt wird.
JFS
4
Was ist, wenn ich den Unterprozess nicht herunterfahren kann, z. aufgrund von Ausnahmen? Der stdout-reader-Thread stirbt nicht und Python hängt, selbst wenn der Haupt-Thread beendet wird, nicht wahr? Wie könnte man das umgehen? Python 2.x unterstützt das Beenden der Threads nicht. Was noch schlimmer ist, unterstützt das Unterbrechen der Threads nicht. :( (offensichtlich sollte man die Ausnahmen behandeln, um sicherzustellen, dass der Unterprozess heruntergefahren wird, aber nur für den Fall, dass dies nicht der Fall ist, was können Sie tun?)
n611x007
3
Ich habe einige freundliche Wrapper davon im Paket shelljob pypi.python.org/pypi/shelljob
edA-qa mort-ora-y
77

Ich hatte oft ein ähnliches Problem; Python-Programme, die ich häufig schreibe, müssen in der Lage sein, einige primäre Funktionen auszuführen und gleichzeitig Benutzereingaben über die Befehlszeile (stdin) zu akzeptieren. Das einfache Einfügen der Funktionen zur Verarbeitung von Benutzereingaben in einen anderen Thread löst das Problem nicht, da es readline()blockiert und keine Zeitüberschreitung aufweist. Wenn die primäre Funktionalität vollständig ist und nicht mehr auf weitere Benutzereingaben gewartet werden muss, möchte ich normalerweise, dass mein Programm beendet wird, dies ist jedoch nicht möglich, da readline()der andere Thread immer noch blockiert und auf eine Zeile wartet. Eine Lösung, die ich für dieses Problem gefunden habe, besteht darin, stdin mit dem Modul fcntl zu einer nicht blockierenden Datei zu machen:

import fcntl
import os
import sys

# make stdin a non-blocking file
fd = sys.stdin.fileno()
fl = fcntl.fcntl(fd, fcntl.F_GETFL)
fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)

# user input handling thread
while mainThreadIsRunning:
      try: input = sys.stdin.readline()
      except: continue
      handleInput(input)

Meiner Meinung nach ist dies etwas sauberer als die Verwendung der Select- oder Signalmodule, um dieses Problem zu lösen, aber es funktioniert auch nur unter UNIX ...

Jesse
quelle
1
Gemäß den Dokumenten kann fcntl () entweder einen Dateideskriptor oder ein Objekt mit der Methode .fileno () empfangen.
Denilson Sá Maia
10
Jesses Antwort ist nicht richtig. Laut Guido funktioniert readline im nicht blockierenden Modus nicht korrekt und nicht vor Python 3000. bugs.python.org/issue1175#msg56041 Wenn Sie fcntl verwenden möchten, um die Datei in den nicht blockierenden Modus zu versetzen, Sie müssen das untergeordnete os.read () verwenden und die Zeilen selbst trennen. Das Mischen von fcntl mit Aufrufen auf hoher Ebene, die eine Zeilenpufferung durchführen, ist problematisch.
Anonnn
2
Die Verwendung von readline scheint in Python 2 falsch zu sein. Siehe anonnns Antwort stackoverflow.com/questions/375427/…
Catalin Iacob
10
Bitte verwenden Sie keine Besetztschleifen. Verwenden Sie poll () mit einer Zeitüberschreitung, um auf die Daten zu warten.
Ivo Danihelka
@Stefano was ist buffer_sizedefiniert als?
Katze
39

Python 3.4 führt eine neue vorläufige API für das asynchrone E / A- asyncioModul ein .

Der Ansatz ähnelt der twistedAntwort von @Bryan Ward. Definieren Sie ein Protokoll, und seine Methoden werden aufgerufen, sobald die Daten bereit sind:

#!/usr/bin/env python3
import asyncio
import os

class SubprocessProtocol(asyncio.SubprocessProtocol):
    def pipe_data_received(self, fd, data):
        if fd == 1: # got stdout data (bytes)
            print(data)

    def connection_lost(self, exc):
        loop.stop() # end loop.run_forever()

if os.name == 'nt':
    loop = asyncio.ProactorEventLoop() # for subprocess' pipes on Windows
    asyncio.set_event_loop(loop)
else:
    loop = asyncio.get_event_loop()
try:
    loop.run_until_complete(loop.subprocess_exec(SubprocessProtocol, 
        "myprogram.exe", "arg1", "arg2"))
    loop.run_forever()
finally:
    loop.close()

Siehe "Unterprozess" in den Dokumenten .

Es gibt eine übergeordnete Schnittstelle asyncio.create_subprocess_exec(), die ProcessObjekte zurückgibt , mit denen eine Zeile mithilfe von StreamReader.readline()Coroutine (mit async/ awaitPython 3.5+ -Syntax ) asynchron gelesen werden kann :

#!/usr/bin/env python3.5
import asyncio
import locale
import sys
from asyncio.subprocess import PIPE
from contextlib import closing

async def readline_and_kill(*args):
    # start child process
    process = await asyncio.create_subprocess_exec(*args, stdout=PIPE)

    # read line (sequence of bytes ending with b'\n') asynchronously
    async for line in process.stdout:
        print("got line:", line.decode(locale.getpreferredencoding(False)))
        break
    process.kill()
    return await process.wait() # wait for the child process to exit


if sys.platform == "win32":
    loop = asyncio.ProactorEventLoop()
    asyncio.set_event_loop(loop)
else:
    loop = asyncio.get_event_loop()

with closing(loop):
    sys.exit(loop.run_until_complete(readline_and_kill(
        "myprogram.exe", "arg1", "arg2")))

readline_and_kill() führt folgende Aufgaben aus:

  • Starten Sie den Unterprozess und leiten Sie die Standardausgabe in eine Pipe um
  • Lesen Sie asynchron eine Zeile aus dem Unterprozess 'stdout
  • Unterprozess beenden
  • Warten Sie, bis es beendet ist

Jeder Schritt kann bei Bedarf durch Timeout-Sekunden begrenzt werden.

jfs
quelle
Wenn ich so etwas mit Python 3.4-Coroutinen versuche, erhalte ich erst eine Ausgabe, wenn das gesamte Skript ausgeführt wurde. Ich möchte, dass eine Ausgabezeile gedruckt wird, sobald der Unterprozess eine Zeile druckt. Folgendes habe ich: pastebin.com/qPssFGep .
Flutefreak7
1
@ flutefreak7: Pufferung Probleme auf die aktuelle Frage nicht verwandt sind. Folgen Sie dem Link für mögliche Lösungen.
JFS
Vielen Dank! Das Problem für mein Skript wurde durch einfaches Verwenden gelöst, print(text, flush=True)sodass der gedruckte Text dem anrufenden Beobachter sofort zur Verfügung steht readline. Als ich es mit der Fortran-basierten ausführbaren Datei getestet habe, die ich eigentlich umschließen / beobachten möchte, puffert es die Ausgabe nicht und verhält sich daher wie erwartet.
Flutefreak7
Ist es möglich, den Unterprozess bestehen zu lassen und weitere Lese- / Schreibvorgänge auszuführen? readline_and_killfunktioniert in Ihrem zweiten Skript sehr ähnlich subprocess.comunicate, da es den Prozess nach einem Lese- / Schreibvorgang beendet. Ich sehe auch, dass Sie eine einzelne Pipe verwenden stdout, die der Unterprozess als nicht blockierend behandelt. Der Versuch , beide zu verwenden , stdoutund stderr ich finde , ich Blockierung am Ende .
Carel
@Carel Der Code in der Antwort funktioniert wie beabsichtigt, wie in der Antwort explizit beschrieben. Falls gewünscht, ist es möglich, ein anderes Verhalten zu implementieren. Beide Pipes blockieren bei Verwendung gleichermaßen nicht. Hier ist ein Beispiel für das gleichzeitige Lesen beider Pipes .
JFS
19

Probieren Sie das asyncproc- Modul aus. Zum Beispiel:

import os
from asyncproc import Process
myProc = Process("myprogram.app")

while True:
    # check to see if process has ended
    poll = myProc.wait(os.WNOHANG)
    if poll != None:
        break
    # print any new output
    out = myProc.read()
    if out != "":
        print out

Das Modul kümmert sich um das gesamte Einfädeln, wie von S.Lott vorgeschlagen.

Noah
quelle
1
Absolut brilliant. Viel einfacher als das rohe Unterprozessmodul. Funktioniert perfekt für mich unter Ubuntu.
Cerin
12
asyncproc funktioniert nicht unter Windows und Windows unterstützt os.WNOHANG nicht :-(
Bryan Oakley
26
asyncproc ist GPL, was seine Verwendung weiter einschränkt :-(
Bryan Oakley
Vielen Dank. Eine kleine Sache: Es scheint, dass das Ersetzen von Tabulatoren durch 8 Leerzeichen in asyncproc.py der richtige Weg ist :)
benjaoming
Es sieht nicht so aus, als könnten Sie den Rückkehrcode des Prozesses, den Sie gestartet haben, über das asyncproc-Modul abrufen. nur die Ausgabe, die es generiert hat.
Grayaii
17

Sie können dies wirklich einfach in Twisted tun . Abhängig von Ihrer vorhandenen Codebasis ist dies möglicherweise nicht so einfach zu verwenden. Wenn Sie jedoch eine verdrehte Anwendung erstellen, werden solche Dinge fast trivial. Sie erstellen eine ProcessProtocolKlasse und überschreiben die outReceived()Methode. Twisted (abhängig vom verwendeten Reaktor) ist normalerweise nur eine große select()Schleife mit installierten Rückrufen, um Daten von verschiedenen Dateideskriptoren (häufig Netzwerk-Sockets) zu verarbeiten. Die outReceived()Methode installiert also einfach einen Rückruf für die Verarbeitung von Daten, die von kommen STDOUT. Ein einfaches Beispiel für dieses Verhalten lautet wie folgt:

from twisted.internet import protocol, reactor

class MyProcessProtocol(protocol.ProcessProtocol):

    def outReceived(self, data):
        print data

proc = MyProcessProtocol()
reactor.spawnProcess(proc, './myprogram', ['./myprogram', 'arg1', 'arg2', 'arg3'])
reactor.run()

Die Twisted-Dokumentation enthält einige gute Informationen dazu.

Wenn Sie Ihre gesamte Anwendung auf Twisted aufbauen, wird die asynchrone Kommunikation mit anderen lokalen oder Remote-Prozessen so elegant. Wenn Ihr Programm jedoch nicht auf Twisted basiert, ist dies nicht wirklich hilfreich. Hoffentlich kann dies für andere Leser hilfreich sein, auch wenn es für Ihre spezielle Anwendung nicht gilt.

Bryan Ward
quelle
nicht gut. selectsollte nicht unter Windows mit Dateideskriptoren funktionieren, laut docs
n611x007
2
@naxa Ich glaube nicht, dass das select(), auf das er sich bezieht, dasselbe ist, das du bist. Ich gehe davon aus, weil es Twistedunter Windows funktioniert ...
notbad.jpeg
1
"Verdrillt (abhängig vom verwendeten Reaktor) ist normalerweise nur eine große select () - Schleife" bedeutet, dass mehrere Reaktoren zur Auswahl stehen. Der select()ist der portabelste unter Unixen und Unix-Likes, aber es gibt auch zwei Reaktoren für Windows: twistedmatrix.com/documents/current/core/howto/…
clacke
14

Verwenden Sie select & read (1).

import subprocess     #no new requirements
def readAllSoFar(proc, retVal=''): 
  while (select.select([proc.stdout],[],[],0)[0]!=[]):   
    retVal+=proc.stdout.read(1)
  return retVal
p = subprocess.Popen(['/bin/ls'], stdout=subprocess.PIPE)
while not p.poll():
  print (readAllSoFar(p))

Für readline () - wie:

lines = ['']
while not p.poll():
  lines = readAllSoFar(p, lines[-1]).split('\n')
  for a in range(len(lines)-1):
    print a
lines = readAllSoFar(p, lines[-1]).split('\n')
for a in range(len(lines)-1):
  print a
Andy Jackson
quelle
6
nicht gut. selectsollte nicht unter Windows mit Dateideskriptoren funktionieren, laut docs
n611x007
OH MEIN GOTT. Lesen Sie Megabyte oder möglicherweise Gigabyte Zeichen für Zeichen ... das ist die schlimmste Idee, die ich seit langem gesehen habe ... Unnötig zu erwähnen, dass dieser Code nicht funktioniert, denn proc.stdout.read()egal wie klein das Argument ist ein blockierender Anruf.
wvxvw
OSError: [WinError 10093] Either the application has not called WSAStartup, or WSAStartup failed
nmz787
8

Eine Lösung besteht darin, einen anderen Prozess zu erstellen, um das Lesen des Prozesses durchzuführen, oder einen Thread des Prozesses mit einer Zeitüberschreitung zu erstellen.

Hier ist die Thread-Version einer Timeout-Funktion:

http://code.activestate.com/recipes/473878/

Müssen Sie jedoch den Standard lesen, wenn er eingeht? Eine andere Lösung könnte darin bestehen, die Ausgabe in eine Datei zu kopieren und mit p.wait () zu warten, bis der Prozess abgeschlossen ist .

f = open('myprogram_output.txt','w')
p = subprocess.Popen('myprogram.exe', stdout=f)
p.wait()
f.close()


str = open('myprogram_output.txt','r').read()
Mönch
quelle
Es scheint, als würde der Thread von recpie nach einer Zeitüberschreitung nicht beendet und das Beenden hängt davon ab, ob der Teilprozess (sg. ansonsten in dieser Hinsicht nicht verwandt) abgebrochen werden kann (eine Sache, die Sie können sollten, aber nur für den Fall, dass Sie nicht können ..). .
n611x007
7

Haftungsausschluss: Dies funktioniert nur für Tornados

Sie können dies tun, indem Sie den fd so einstellen, dass er nicht blockiert, und dann ioloop verwenden, um Rückrufe zu registrieren. Ich habe dies in einem Ei namens tornado_subprocess verpackt und Sie können es über PyPI installieren:

easy_install tornado_subprocess

Jetzt können Sie so etwas tun:

import tornado_subprocess
import tornado.ioloop

    def print_res( status, stdout, stderr ) :
    print status, stdout, stderr
    if status == 0:
        print "OK:"
        print stdout
    else:
        print "ERROR:"
        print stderr

t = tornado_subprocess.Subprocess( print_res, timeout=30, args=[ "cat", "/etc/passwd" ] )
t.start()
tornado.ioloop.IOLoop.instance().start()

Sie können es auch mit einem RequestHandler verwenden

class MyHandler(tornado.web.RequestHandler):
    def on_done(self, status, stdout, stderr):
        self.write( stdout )
        self.finish()

    @tornado.web.asynchronous
    def get(self):
        t = tornado_subprocess.Subprocess( self.on_done, timeout=30, args=[ "cat", "/etc/passwd" ] )
        t.start()
Vukasin Toroman
quelle
Danke für die nette Funktion! Nur um zu verdeutlichen, warum können wir nicht einfach threading.Threadneue nicht blockierende Prozesse erstellen? Ich habe es in on_messageeiner Tornado-Websocket-Instanz verwendet, und es hat den Job gut gemacht.
VisioN
1
Beim Tornado wird meistens vom Einfädeln abgeraten. Sie eignen sich gut für kleine, kurz laufende Funktionen. Sie können hier darüber lesen: stackoverflow.com/questions/7846323/tornado-web-and-threads github.com/facebook/tornado/wiki/Threading-and-concurrency
Vukasin Toroman
@ VukasinToroman du hast mich hier wirklich damit gerettet. Vielen Dank für das Modul tornado_subprocess :)
James Gentes
funktioniert das unter Windows? (Beachten Sie, dass selectmit Dateideskriptoren nicht )
n611x007
Diese Bibliothek verwendet den selectAufruf nicht. Ich habe dies unter Windows nicht versucht, aber Sie würden wahrscheinlich auf Probleme stoßen, da die Bibliothek das fcntlModul verwendet. Kurz gesagt: Nein, dies wird unter Windows wahrscheinlich nicht funktionieren.
Vukasin Toroman
6

Bestehende Lösungen haben bei mir nicht funktioniert (Details unten). Was schließlich funktionierte, war die Implementierung von readline mit read (1) (basierend auf dieser Antwort ). Letzteres blockiert nicht:

from subprocess import Popen, PIPE
from threading import Thread
def process_output(myprocess): #output-consuming thread
    nextline = None
    buf = ''
    while True:
        #--- extract line using read(1)
        out = myprocess.stdout.read(1)
        if out == '' and myprocess.poll() != None: break
        if out != '':
            buf += out
            if out == '\n':
                nextline = buf
                buf = ''
        if not nextline: continue
        line = nextline
        nextline = None

        #--- do whatever you want with line here
        print 'Line is:', line
    myprocess.stdout.close()

myprocess = Popen('myprogram.exe', stdout=PIPE) #output-producing process
p1 = Thread(target=process_output, args=(dcmpid,)) #output-consuming thread
p1.daemon = True
p1.start()

#--- do whatever here and then kill process and thread if needed
if myprocess.poll() == None: #kill process; will automatically stop thread
    myprocess.kill()
    myprocess.wait()
if p1 and p1.is_alive(): #wait for thread to finish
    p1.join()

Warum vorhandene Lösungen nicht funktionierten:

  1. Lösungen, die Readline erfordern (einschließlich der auf Warteschlangen basierenden), blockieren immer. Es ist schwierig (unmöglich?), Den Thread zu beenden, der readline ausführt. Es wird nur beendet, wenn der Prozess, der es erstellt hat, abgeschlossen ist, nicht jedoch, wenn der ausgabeproduzierende Prozess beendet wird.
  2. Das Mischen von fcntl auf niedriger Ebene mit Readline-Aufrufen auf hoher Ebene funktioniert möglicherweise nicht richtig, wie anonnn hervorgehoben hat.
  3. Die Verwendung von select.poll () ist ordentlich, funktioniert aber laut Python-Dokumenten unter Windows nicht.
  4. Die Verwendung von Bibliotheken von Drittanbietern scheint für diese Aufgabe übertrieben und fügt zusätzliche Abhängigkeiten hinzu.
Vikram Pudi
quelle
1
1. q.get_nowait()von meiner Antwort darf niemals blockieren, das ist der Sinn der Verwendung. 2. Der Thread, der readline ( enqueue_output()function ) ausführt, wird auf EOF beendet, z. B. einschließlich des Falls, in dem der ausgabeproduzierende Prozess beendet wird. Wenn Sie glauben, dass es nicht so ist; Bitte geben Sie ein vollständiges Beispiel für einen minimalen Code an , das etwas anderes anzeigt (möglicherweise als neue Frage ).
JFS
1
@sebastian Ich habe eine Stunde oder länger damit verbracht, ein minimales Beispiel zu finden. Am Ende muss ich zustimmen, dass Ihre Antwort alle Fälle behandelt. Ich denke, es hat früher bei mir nicht funktioniert, denn als ich versuchte, den Output-produzierenden Prozess zu beenden, wurde es bereits beendet und es gab einen schwer zu debuggenden Fehler. Die Stunde war gut angelegt, denn während ich ein minimales Beispiel fand, konnte ich eine einfachere Lösung finden.
Vikram Pudi
Könnten Sie auch die einfachere Lösung posten? :) (wenn es anders ist als bei Sebastian)
n611x007
@ Gefahr89: Ich denke dcmpid = myprocess.
ViFI
In Bedingung nach dem Aufruf von read () (kurz nach True): out wird niemals eine leere Zeichenfolge sein, da Sie mindestens Zeichenfolge / Bytes mit einer Länge von 1
lesen.
6

Hier ist mein Code, der verwendet wird, um jede Ausgabe des Unterprozesses so schnell wie möglich abzufangen, einschließlich Teilzeilen. Es pumpt gleichzeitig und stdout und stderr in fast korrekter Reihenfolge.

Getestet und korrekt unter Python 2.7 Linux & Windows.

#!/usr/bin/python
#
# Runner with stdout/stderr catcher
#
from sys import argv
from subprocess import Popen, PIPE
import os, io
from threading import Thread
import Queue
def __main__():
    if (len(argv) > 1) and (argv[-1] == "-sub-"):
        import time, sys
        print "Application runned!"
        time.sleep(2)
        print "Slept 2 second"
        time.sleep(1)
        print "Slept 1 additional second",
        time.sleep(2)
        sys.stderr.write("Stderr output after 5 seconds")
        print "Eol on stdin"
        sys.stderr.write("Eol on stderr\n")
        time.sleep(1)
        print "Wow, we have end of work!",
    else:
        os.environ["PYTHONUNBUFFERED"]="1"
        try:
            p = Popen( argv + ["-sub-"],
                       bufsize=0, # line-buffered
                       stdin=PIPE, stdout=PIPE, stderr=PIPE )
        except WindowsError, W:
            if W.winerror==193:
                p = Popen( argv + ["-sub-"],
                           shell=True, # Try to run via shell
                           bufsize=0, # line-buffered
                           stdin=PIPE, stdout=PIPE, stderr=PIPE )
            else:
                raise
        inp = Queue.Queue()
        sout = io.open(p.stdout.fileno(), 'rb', closefd=False)
        serr = io.open(p.stderr.fileno(), 'rb', closefd=False)
        def Pump(stream, category):
            queue = Queue.Queue()
            def rdr():
                while True:
                    buf = stream.read1(8192)
                    if len(buf)>0:
                        queue.put( buf )
                    else:
                        queue.put( None )
                        return
            def clct():
                active = True
                while active:
                    r = queue.get()
                    try:
                        while True:
                            r1 = queue.get(timeout=0.005)
                            if r1 is None:
                                active = False
                                break
                            else:
                                r += r1
                    except Queue.Empty:
                        pass
                    inp.put( (category, r) )
            for tgt in [rdr, clct]:
                th = Thread(target=tgt)
                th.setDaemon(True)
                th.start()
        Pump(sout, 'stdout')
        Pump(serr, 'stderr')

        while p.poll() is None:
            # App still working
            try:
                chan,line = inp.get(timeout = 1.0)
                if chan=='stdout':
                    print "STDOUT>>", line, "<?<"
                elif chan=='stderr':
                    print " ERROR==", line, "=?="
            except Queue.Empty:
                pass
        print "Finish"

if __name__ == '__main__':
    __main__()
Datacompboy
quelle
Eine der wenigen Antworten, mit denen Sie Dinge lesen können, die nicht unbedingt mit einem Zeilenumbruch enden.
Totaam
5

Ich füge dieses Problem hinzu, um einen Unterprozess zu lesen. Öffnen Sie stdout. Hier ist meine nicht blockierende Leselösung:

import fcntl

def non_block_read(output):
    fd = output.fileno()
    fl = fcntl.fcntl(fd, fcntl.F_GETFL)
    fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)
    try:
        return output.read()
    except:
        return ""

# Use example
from subprocess import *
sb = Popen("echo test && sleep 1000", shell=True, stdout=PIPE)
sb.kill()

# sb.stdout.read() # <-- This will block
non_block_read(sb.stdout)
'test\n'
Sebastien Claeys
quelle
5
fcntl funktioniert laut Dokumentation nicht unter Windows .
n611x007
@anatolytechtonik verwenden msvcrt.kbhit()stattdessen
Katze
4

Diese Version des nicht blockierenden Lesens erfordert keine speziellen Module und funktioniert bei den meisten Linux-Distributionen sofort.

import os
import sys
import time
import fcntl
import subprocess

def async_read(fd):
    # set non-blocking flag while preserving old flags
    fl = fcntl.fcntl(fd, fcntl.F_GETFL)
    fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)
    # read char until EOF hit
    while True:
        try:
            ch = os.read(fd.fileno(), 1)
            # EOF
            if not ch: break                                                                                                                                                              
            sys.stdout.write(ch)
        except OSError:
            # waiting for data be available on fd
            pass

def shell(args, async=True):
    # merge stderr and stdout
    proc = subprocess.Popen(args, shell=False, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
    if async: async_read(proc.stdout)
    sout, serr = proc.communicate()
    return (sout, serr)

if __name__ == '__main__':
    cmd = 'ping 8.8.8.8'
    sout, serr = shell(cmd.split())
Tom Lime
quelle
3

Hier ist eine einfache Lösung basierend auf Threads, die:

  • funktioniert sowohl unter Linux als auch unter Windows (ohne sich darauf zu verlassen) select ).
  • liest beide stdoutundstderr asynchron.
  • ist nicht auf aktives Polling mit beliebiger Wartezeit angewiesen (CPU-freundlich).
  • wird nicht verwendet asyncio (was zu Konflikten mit anderen Bibliotheken führen kann).
  • wird ausgeführt, bis der untergeordnete Prozess beendet wird.

Drucker.py

import time
import sys

sys.stdout.write("Hello\n")
sys.stdout.flush()
time.sleep(1)
sys.stdout.write("World!\n")
sys.stdout.flush()
time.sleep(1)
sys.stderr.write("That's an error\n")
sys.stderr.flush()
time.sleep(2)
sys.stdout.write("Actually, I'm fine\n")
sys.stdout.flush()
time.sleep(1)

reader.py

import queue
import subprocess
import sys
import threading


def enqueue_stream(stream, queue, type):
    for line in iter(stream.readline, b''):
        queue.put(str(type) + line.decode('utf-8'))
    stream.close()


def enqueue_process(process, queue):
    process.wait()
    queue.put('x')


p = subprocess.Popen('python printer.py', stdout=subprocess.PIPE, stderr=subprocess.PIPE)
q = queue.Queue()
to = threading.Thread(target=enqueue_stream, args=(p.stdout, q, 1))
te = threading.Thread(target=enqueue_stream, args=(p.stderr, q, 2))
tp = threading.Thread(target=enqueue_process, args=(p, q))
te.start()
to.start()
tp.start()

while True:
    line = q.get()
    if line[0] == 'x':
        break
    if line[0] == '2':  # stderr
        sys.stdout.write("\033[0;31m")  # ANSI red color
    sys.stdout.write(line[1:])
    if line[0] == '2':
        sys.stdout.write("\033[0m")  # reset ANSI code
    sys.stdout.flush()

tp.join()
to.join()
te.join()
Olivier Michel
quelle
2

Fügen Sie diese Antwort hier hinzu, da nicht blockierende Pipes unter Windows und Unix festgelegt werden können.

Alle ctypesDetails sind der Antwort von @ techtonik zu verdanken .

Es gibt eine leicht modifizierte Version, die sowohl auf Unix- als auch auf Windows-Systemen verwendet werden kann.

  • Python3-kompatibel (nur geringfügige Änderungen erforderlich) .
  • Enthält die Posix-Version und definiert die Ausnahme, die für beide verwendet werden soll.

Auf diese Weise können Sie dieselbe Funktion und Ausnahme für Unix- und Windows-Code verwenden.

# pipe_non_blocking.py (module)
"""
Example use:

    p = subprocess.Popen(
            command,
            stdout=subprocess.PIPE,
            )

    pipe_non_blocking_set(p.stdout.fileno())

    try:
        data = os.read(p.stdout.fileno(), 1)
    except PortableBlockingIOError as ex:
        if not pipe_non_blocking_is_error_blocking(ex):
            raise ex
"""


__all__ = (
    "pipe_non_blocking_set",
    "pipe_non_blocking_is_error_blocking",
    "PortableBlockingIOError",
    )

import os


if os.name == "nt":
    def pipe_non_blocking_set(fd):
        # Constant could define globally but avoid polluting the name-space
        # thanks to: /programming/34504970
        import msvcrt

        from ctypes import windll, byref, wintypes, WinError, POINTER
        from ctypes.wintypes import HANDLE, DWORD, BOOL

        LPDWORD = POINTER(DWORD)

        PIPE_NOWAIT = wintypes.DWORD(0x00000001)

        def pipe_no_wait(pipefd):
            SetNamedPipeHandleState = windll.kernel32.SetNamedPipeHandleState
            SetNamedPipeHandleState.argtypes = [HANDLE, LPDWORD, LPDWORD, LPDWORD]
            SetNamedPipeHandleState.restype = BOOL

            h = msvcrt.get_osfhandle(pipefd)

            res = windll.kernel32.SetNamedPipeHandleState(h, byref(PIPE_NOWAIT), None, None)
            if res == 0:
                print(WinError())
                return False
            return True

        return pipe_no_wait(fd)

    def pipe_non_blocking_is_error_blocking(ex):
        if not isinstance(ex, PortableBlockingIOError):
            return False
        from ctypes import GetLastError
        ERROR_NO_DATA = 232

        return (GetLastError() == ERROR_NO_DATA)

    PortableBlockingIOError = OSError
else:
    def pipe_non_blocking_set(fd):
        import fcntl
        fl = fcntl.fcntl(fd, fcntl.F_GETFL)
        fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)
        return True

    def pipe_non_blocking_is_error_blocking(ex):
        if not isinstance(ex, PortableBlockingIOError):
            return False
        return True

    PortableBlockingIOError = BlockingIOError

Um das Lesen unvollständiger Daten zu vermeiden, habe ich meinen eigenen Readline-Generator geschrieben (der die Byte-Zeichenfolge für jede Zeile zurückgibt).

Es ist ein Generator, so dass Sie zum Beispiel ...

def non_blocking_readlines(f, chunk=1024):
    """
    Iterate over lines, yielding b'' when nothings left
    or when new data is not yet available.

    stdout_iter = iter(non_blocking_readlines(process.stdout))

    line = next(stdout_iter)  # will be a line or b''.
    """
    import os

    from .pipe_non_blocking import (
            pipe_non_blocking_set,
            pipe_non_blocking_is_error_blocking,
            PortableBlockingIOError,
            )

    fd = f.fileno()
    pipe_non_blocking_set(fd)

    blocks = []

    while True:
        try:
            data = os.read(fd, chunk)
            if not data:
                # case were reading finishes with no trailing newline
                yield b''.join(blocks)
                blocks.clear()
        except PortableBlockingIOError as ex:
            if not pipe_non_blocking_is_error_blocking(ex):
                raise ex

            yield b''
            continue

        while True:
            n = data.find(b'\n')
            if n == -1:
                break

            yield b''.join(blocks) + data[:n + 1]
            data = data[n + 1:]
            blocks.clear()
        blocks.append(data)
ideasman42
quelle
(1) Dieser Kommentar weist darauf hin, dass readline()er mit nicht blockierenden Pipes (wie z. B. set using fcntl) unter Python 2 nicht funktioniert. Glauben Sie, dass er nicht mehr korrekt ist? (Meine Antwort enthält den Link ( fcntl), der die gleichen Informationen enthält, aber jetzt gelöscht zu sein scheint.) (2) Sehen Sie, wie multiprocessing.connection.PipeverwendetSetNamedPipeHandleState
jfs
Ich habe das nur auf Python3 getestet. Aber habe diese Informationen auch gesehen und erwarte, dass sie gültig bleiben. Ich habe auch meinen eigenen Code geschrieben, um anstelle von readline zu verwenden. Ich habe meine Antwort aktualisiert, um sie einzuschließen.
ideasman42
2

Ich habe das Problem des ursprünglichen Fragestellers, wollte aber keine Threads aufrufen. Ich mischte Jesses Lösung mit einem direkten read () aus der Pipe und meinem eigenen Buffer-Handler für Zeilenlesevorgänge (mein Unterprozess - ping - schrieb jedoch immer vollständige Zeilen <eine Systemseitengröße). Ich vermeide das Warten, indem ich nur eine von einem Objekt registrierte Io-Uhr einlese. Heutzutage führe ich normalerweise Code innerhalb eines Gobjects MainLoop aus, um Threads zu vermeiden.

def set_up_ping(ip, w):
# run the sub-process
# watch the resultant pipe
p = subprocess.Popen(['/bin/ping', ip], stdout=subprocess.PIPE)
# make stdout a non-blocking file
fl = fcntl.fcntl(p.stdout, fcntl.F_GETFL)
fcntl.fcntl(p.stdout, fcntl.F_SETFL, fl | os.O_NONBLOCK)
stdout_gid = gobject.io_add_watch(p.stdout, gobject.IO_IN, w)
return stdout_gid # for shutting down

Der Beobachter ist

def watch(f, *other):
print 'reading',f.read()
return True

Das Hauptprogramm richtet einen Ping ein und ruft dann die Gobject-Mail-Schleife auf.

def main():
set_up_ping('192.168.1.8', watch)
# discard gid as unused here
gobject.MainLoop().run()

Alle anderen Arbeiten sind mit Rückrufen in gobject verbunden.

Dave Kitchen
quelle
2

In modernem Python sieht es viel besser aus.

Hier ist ein einfaches Kinderprogramm, "hello.py":

#!/usr/bin/env python3

while True:
    i = input()
    if i == "quit":
        break
    print(f"hello {i}")

Und ein Programm, um damit zu interagieren:

import asyncio


async def main():
    proc = await asyncio.subprocess.create_subprocess_exec(
        "./hello.py", stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE
    )
    proc.stdin.write(b"bob\n")
    print(await proc.stdout.read(1024))
    proc.stdin.write(b"alice\n")
    print(await proc.stdout.read(1024))
    proc.stdin.write(b"quit\n")
    await proc.wait()


asyncio.run(main())

Das druckt aus:

b'hello bob\n'
b'hello alice\n'

Beachten Sie, dass das eigentliche Muster, das auch hier und in verwandten Fragen von fast allen vorherigen Antworten verwendet wird, darin besteht, den stdout-Dateideskriptor des Kindes auf nicht blockierend zu setzen und ihn dann in einer Art Auswahlschleife abzufragen. Heutzutage wird diese Schleife natürlich von asyncio bereitgestellt.

user240515
quelle
1

Die Auswahl Sie feststellen, wo sich die nächste nützliche Eingabe befindet.

Mit separaten Threads sind Sie jedoch fast immer zufriedener. Einer blockiert das stdin, ein anderer tut, wo immer Sie nicht blockiert werden möchten.

S.Lott
quelle
11
Ich denke, diese Antwort ist aus zwei Gründen nicht hilfreich: (a) Das Auswahlmodul funktioniert nicht mit Pipes unter Windows (wie der bereitgestellte Link eindeutig angibt), was die Absichten des OP, eine tragbare Lösung zu haben, zunichte macht. (b) Asynchrone Threads ermöglichen keinen synchronen Dialog zwischen dem übergeordneten und dem untergeordneten Prozess. Was ist, wenn der übergeordnete Prozess die nächste Aktion gemäß der nächsten vom Kind gelesenen Zeile auslösen möchte?!
ThomasH
4
select ist auch nicht nützlich, da Pythons Lesevorgänge auch nach der Auswahl blockiert werden, da es keine Standard-C-Semantik hat und keine Teildaten zurückgibt.
Helmut Grohne
Ein separater Thresd zum Lesen aus der Ausgabe des Kindes löste mein ähnliches Problem. Wenn Sie eine synchrone Interaktion benötigen, können Sie diese Lösung vermutlich nicht verwenden (es sei denn, Sie wissen, welche Ausgabe zu erwarten ist). Ich hätte diese Antwort akzeptiert
Emiliano
1

Warum Thread & Queue stören? Im Gegensatz zu readline () blockiert BufferedReader.read1 () das Warten auf \ r \ n nicht und gibt so schnell wie möglich zurück, wenn eine Ausgabe eingeht.

#!/usr/bin/python
from subprocess import Popen, PIPE, STDOUT
import io

def __main__():
    try:
        p = Popen( ["ping", "-n", "3", "127.0.0.1"], stdin=PIPE, stdout=PIPE, stderr=STDOUT )
    except: print("Popen failed"); quit()
    sout = io.open(p.stdout.fileno(), 'rb', closefd=False)
    while True:
        buf = sout.read1(1024)
        if len(buf) == 0: break
        print buf,

if __name__ == '__main__':
    __main__()
mfmain
quelle
Wird es so schnell wie möglich zurückgegeben, wenn nichts eingeht? Wenn dies nicht der Fall ist, blockiert es.
Mathieu Pagé
@ MathieuPagé ist richtig. read1wird blockiert, wenn die ersten zugrunde liegenden Leseblöcke, was passiert, wenn die Pipe noch offen ist, aber keine Eingabe verfügbar ist.
Jack O'Connor
1

In meinem Fall benötigte ich ein Protokollierungsmodul, das die Ausgabe der Hintergrundanwendungen abfängt und erweitert (Hinzufügen von Zeitstempeln, Farben usw.).

Am Ende hatte ich einen Hintergrund-Thread, der die eigentliche E / A ausführt. Der folgende Code gilt nur für POSIX-Plattformen. Ich habe nicht wesentliche Teile entfernt.

Wenn jemand dieses Biest für längere Zeit verwenden wird, sollten Sie offene Deskriptoren verwalten. In meinem Fall war es kein großes Problem.

# -*- python -*-
import fcntl
import threading
import sys, os, errno
import subprocess

class Logger(threading.Thread):
    def __init__(self, *modules):
        threading.Thread.__init__(self)
        try:
            from select import epoll, EPOLLIN
            self.__poll = epoll()
            self.__evt = EPOLLIN
            self.__to = -1
        except:
            from select import poll, POLLIN
            print 'epoll is not available'
            self.__poll = poll()
            self.__evt = POLLIN
            self.__to = 100
        self.__fds = {}
        self.daemon = True
        self.start()

    def run(self):
        while True:
            events = self.__poll.poll(self.__to)
            for fd, ev in events:
                if (ev&self.__evt) != self.__evt:
                    continue
                try:
                    self.__fds[fd].run()
                except Exception, e:
                    print e

    def add(self, fd, log):
        assert not self.__fds.has_key(fd)
        self.__fds[fd] = log
        self.__poll.register(fd, self.__evt)

class log:
    logger = Logger()

    def __init__(self, name):
        self.__name = name
        self.__piped = False

    def fileno(self):
        if self.__piped:
            return self.write
        self.read, self.write = os.pipe()
        fl = fcntl.fcntl(self.read, fcntl.F_GETFL)
        fcntl.fcntl(self.read, fcntl.F_SETFL, fl | os.O_NONBLOCK)
        self.fdRead = os.fdopen(self.read)
        self.logger.add(self.read, self)
        self.__piped = True
        return self.write

    def __run(self, line):
        self.chat(line, nl=False)

    def run(self):
        while True:
            try: line = self.fdRead.readline()
            except IOError, exc:
                if exc.errno == errno.EAGAIN:
                    return
                raise
            self.__run(line)

    def chat(self, line, nl=True):
        if nl: nl = '\n'
        else: nl = ''
        sys.stdout.write('[%s] %s%s' % (self.__name, line, nl))

def system(command, param=[], cwd=None, env=None, input=None, output=None):
    args = [command] + param
    p = subprocess.Popen(args, cwd=cwd, stdout=output, stderr=output, stdin=input, env=env, bufsize=0)
    p.wait()

ls = log('ls')
ls.chat('go')
system("ls", ['-l', '/'], output=ls)

date = log('date')
date.chat('go')
system("date", output=date)
Dmytro
quelle
1

Mein Problem ist etwas anders, da ich sowohl stdout als auch stderr aus einem laufenden Prozess sammeln wollte, aber letztendlich dasselbe, da ich die Ausgabe in einem Widget so rendern wollte, wie sie generiert wurde.

Ich wollte nicht auf viele der vorgeschlagenen Problemumgehungen mit Warteschlangen oder zusätzlichen Threads zurückgreifen, da diese nicht erforderlich sein sollten, um eine so allgemeine Aufgabe wie das Ausführen eines anderen Skripts und das Sammeln seiner Ausgabe auszuführen.

Nachdem ich die vorgeschlagenen Lösungen und Python-Dokumente gelesen hatte, löste ich mein Problem mit der folgenden Implementierung. Ja, es funktioniert nur für POSIX, da ich den selectFunktionsaufruf verwende.

Ich bin damit einverstanden, dass die Dokumente verwirrend sind und die Implementierung für eine solche häufig verwendete Skriptaufgabe umständlich ist. Ich glaube, dass ältere Versionen von Python unterschiedliche Standardeinstellungen Popenund Erklärungen haben, was zu großer Verwirrung geführt hat. Dies scheint sowohl für Python 2.7.12 als auch für 3.5.2 gut zu funktionieren.

Der Schlüssel bestand darin, die bufsize=1Zeilenpufferung festzulegen und dann universal_newlines=Trueals Textdatei anstelle einer Binärdatei zu verarbeiten, die bei der Einstellung als Standard zu gelten scheint bufsize=1.

class workerThread(QThread):
   def __init__(self, cmd):
      QThread.__init__(self)
      self.cmd = cmd
      self.result = None           ## return code
      self.error = None            ## flag indicates an error
      self.errorstr = ""           ## info message about the error

   def __del__(self):
      self.wait()
      DEBUG("Thread removed")

   def run(self):
      cmd_list = self.cmd.split(" ")   
      try:
         cmd = subprocess.Popen(cmd_list, bufsize=1, stdin=None
                                        , universal_newlines=True
                                        , stderr=subprocess.PIPE
                                        , stdout=subprocess.PIPE)
      except OSError:
         self.error = 1
         self.errorstr = "Failed to execute " + self.cmd
         ERROR(self.errorstr)
      finally:
         VERBOSE("task started...")
      import select
      while True:
         try:
            r,w,x = select.select([cmd.stdout, cmd.stderr],[],[])
            if cmd.stderr in r:
               line = cmd.stderr.readline()
               if line != "":
                  line = line.strip()
                  self.emit(SIGNAL("update_error(QString)"), line)
            if cmd.stdout in r:
               line = cmd.stdout.readline()
               if line == "":
                  break
               line = line.strip()
               self.emit(SIGNAL("update_output(QString)"), line)
         except IOError:
            pass
      cmd.wait()
      self.result = cmd.returncode
      if self.result < 0:
         self.error = 1
         self.errorstr = "Task terminated by signal " + str(self.result)
         ERROR(self.errorstr)
         return
      if self.result:
         self.error = 1
         self.errorstr = "exit code " + str(self.result)
         ERROR(self.errorstr)
         return
      return

ERROR, DEBUG und VERBOSE sind einfach Makros, die die Ausgabe an das Terminal drucken.

Diese Lösung ist meiner Meinung nach zu 99,99% effektiv, da sie weiterhin die Blockierungsfunktion readlineverwendet. Wir gehen daher davon aus, dass der Unterprozess gut ist und vollständige Zeilen ausgibt.

Ich freue mich über Feedback zur Verbesserung der Lösung, da ich Python noch nicht kenne.

Brooke Wallace
quelle
In diesem speziellen Fall können Sie stderr = subprocess.STDOUT im Popen-Konstruktor festlegen und die gesamte Ausgabe von cmd.stdout.readline () abrufen.
Aaron
Schönes klares Beispiel. Hatte Probleme mit select.select (), aber das hat es für mich gelöst.
Maharvey67
0

Ausgehend von der Antwort von JF Sebastian und mehreren anderen Quellen habe ich einen einfachen Subprozessmanager zusammengestellt. Es bietet das nicht blockierende Lesen der Anforderung sowie das parallele Ausführen mehrerer Prozesse. Es verwendet keinen betriebssystemspezifischen Aufruf (den ich kenne) und sollte daher überall funktionieren.

Es ist bei pypi erhältlich, also einfach pip install shelljob. Beispiele und vollständige Dokumente finden Sie auf der Projektseite .

edA-qa mort-ora-y
quelle
0

EDIT: Diese Implementierung blockiert immer noch. Verwenden Sie stattdessen die Antwort von JFSebastian .

Ich habe die beste Antwort versucht , aber das zusätzliche Risiko und die Wartung des Thread-Codes waren besorgniserregend.

Beim Durchsehen des io-Moduls (und der Beschränkung auf 2.6) fand ich BufferedReader. Dies ist meine threadlose, nicht blockierende Lösung.

import io
from subprocess import PIPE, Popen

p = Popen(['myprogram.exe'], stdout=PIPE)

SLEEP_DELAY = 0.001

# Create an io.BufferedReader on the file descriptor for stdout
with io.open(p.stdout.fileno(), 'rb', closefd=False) as buffer:
  while p.poll() == None:
      time.sleep(SLEEP_DELAY)
      while '\n' in bufferedStdout.peek(bufferedStdout.buffer_size):
          line = buffer.readline()
          # do stuff with the line

  # Handle any remaining output after the process has ended
  while buffer.peek():
    line = buffer.readline()
    # do stuff with the line
romc
quelle
haben Sie versucht for line in iter(p.stdout.readline, ""): # do stuff with the line? Es ist threadlos (einzelner Thread) und blockiert, wenn Ihr Code blockiert.
JFS
@ jf-sebastian Ja, ich bin schließlich zu deiner Antwort zurückgekehrt. Meine Implementierung wurde immer noch gelegentlich blockiert. Ich werde meine Antwort bearbeiten, um andere zu warnen, diesen Weg nicht zu gehen.
Romc
0

Ich bin kürzlich auf das gleiche Problem gestoßen, bei dem ich jeweils eine Zeile aus dem Stream (Tail Run im Unterprozess) im nicht blockierenden Modus lesen muss. Ich wollte die nächsten Probleme vermeiden: CPU nicht brennen, Stream nicht um ein Byte lesen ( wie readline) usw.

Hier ist meine Implementierung https://gist.github.com/grubberr/5501e1a9760c3eab5e0a Es unterstützt keine Windows (Umfrage), behandelt EOF nicht, aber es funktioniert gut für mich

grubberr
quelle
der Thread-basierte Antwort ist nicht CPU brennen (Sie angeben , willkürlich können timeoutwie in Ihrer Lösung) und .readline()lesen mehr als ein Byte zu einem Zeitpunkt ( bufsize=1Mittel linie -gepufferter (nur relevant für das Schreiben)). Welche anderen Probleme haben Sie gefunden? Nur-Link-Antworten sind nicht sehr nützlich.
JFS
0

Dies ist ein Beispiel für die Ausführung eines interaktiven Befehls im Unterprozess. Der Standardausgang ist mithilfe eines Pseudo-Terminals interaktiv. Sie können auf folgende Adresse verweisen: https://stackoverflow.com/a/43012138/3555925

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import os
import sys
import select
import termios
import tty
import pty
from subprocess import Popen

command = 'bash'
# command = 'docker run -it --rm centos /bin/bash'.split()

# save original tty setting then set it to raw mode
old_tty = termios.tcgetattr(sys.stdin)
tty.setraw(sys.stdin.fileno())

# open pseudo-terminal to interact with subprocess
master_fd, slave_fd = pty.openpty()

# use os.setsid() make it run in a new process group, or bash job control will not be enabled
p = Popen(command,
          preexec_fn=os.setsid,
          stdin=slave_fd,
          stdout=slave_fd,
          stderr=slave_fd,
          universal_newlines=True)

while p.poll() is None:
    r, w, e = select.select([sys.stdin, master_fd], [], [])
    if sys.stdin in r:
        d = os.read(sys.stdin.fileno(), 10240)
        os.write(master_fd, d)
    elif master_fd in r:
        o = os.read(master_fd, 10240)
        if o:
            os.write(sys.stdout.fileno(), o)

# restore tty settings back
termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_tty)
Liao
quelle
0

Diese Lösung verwendet die select Modul, um "verfügbare Daten zu lesen" aus einem E / A-Stream. Diese Funktion blockiert zunächst, bis Daten verfügbar sind, liest dann jedoch nur die verfügbaren Daten und blockiert sie nicht weiter.

Da das selectModul verwendet wird, funktioniert dies nur unter Unix.

Der Code ist vollständig PEP8-kompatibel.

import select


def read_available(input_stream, max_bytes=None):
    """
    Blocks until any data is available, then all available data is then read and returned.
    This function returns an empty string when end of stream is reached.

    Args:
        input_stream: The stream to read from.
        max_bytes (int|None): The maximum number of bytes to read. This function may return fewer bytes than this.

    Returns:
        str
    """
    # Prepare local variables
    input_streams = [input_stream]
    empty_list = []
    read_buffer = ""

    # Initially block for input using 'select'
    if len(select.select(input_streams, empty_list, empty_list)[0]) > 0:

        # Poll read-readiness using 'select'
        def select_func():
            return len(select.select(input_streams, empty_list, empty_list, 0)[0]) > 0

        # Create while function based on parameters
        if max_bytes is not None:
            def while_func():
                return (len(read_buffer) < max_bytes) and select_func()
        else:
            while_func = select_func

        while True:
            # Read single byte at a time
            read_data = input_stream.read(1)
            if len(read_data) == 0:
                # End of stream
                break
            # Append byte to string buffer
            read_buffer += read_data
            # Check if more data is available
            if not while_func():
                break

    # Return read buffer
    return read_buffer
Bradley Odell
quelle
0

Ich habe mich auch dem von Jesse beschriebenen Problem gestellt und es gelöst, indem ich "select" verwendet habe, wie es Bradley , Andy und andere getan haben, aber in einem Blockierungsmodus, um eine Besetztschleife zu vermeiden. Es verwendet eine Dummy-Pipe als Fake-Standard. Die Auswahl blockiert und wartet, bis entweder stdin oder die Pipe fertig sind. Wenn eine Taste gedrückt wird, entsperrt stdin die Auswahl und der Tastenwert kann mit read (1) abgerufen werden. Wenn ein anderer Thread in die Pipe schreibt, entsperrt die Pipe die Auswahl und dies kann als Hinweis darauf angesehen werden, dass die Notwendigkeit für stdin vorbei ist. Hier ist ein Referenzcode:

import sys
import os
from select import select

# -------------------------------------------------------------------------    
# Set the pipe (fake stdin) to simulate a final key stroke
# which will unblock the select statement
readEnd, writeEnd = os.pipe()
readFile = os.fdopen(readEnd)
writeFile = os.fdopen(writeEnd, "w")

# -------------------------------------------------------------------------
def getKey():

    # Wait for stdin or pipe (fake stdin) to be ready
    dr,dw,de = select([sys.__stdin__, readFile], [], [])

    # If stdin is the one ready then read it and return value
    if sys.__stdin__ in dr:
        return sys.__stdin__.read(1)   # For Windows use ----> getch() from module msvcrt

    # Must finish
    else:
        return None

# -------------------------------------------------------------------------
def breakStdinRead():
    writeFile.write(' ')
    writeFile.flush()

# -------------------------------------------------------------------------
# MAIN CODE

# Get key stroke
key = getKey()

# Keyboard input
if key:
    # ... do your stuff with the key value

# Faked keystroke
else:
    # ... use of stdin finished

# -------------------------------------------------------------------------
# OTHER THREAD CODE

breakStdinRead()
gonzaedu61
quelle
HINWEIS: Damit dies unter Windows funktioniert, sollte die Pipe durch einen Sockel ersetzt werden. Ich habe es noch nicht ausprobiert, aber es sollte gemäß der Dokumentation funktionieren.
Gonzaedu61
0

Probieren Sie wexpect aus , die Windows-Alternative von pexpect .

import wexpect

p = wexpect.spawn('myprogram.exe')
p.stdout.readline('.')               // regex pattern of any character
output_str = p.after()
Feindealpfa
quelle
0

Auf Unix-ähnlichen Systemen und Python 3.5+ gibt os.set_blockinges genau das, was es sagt.

import os
import time
import subprocess

cmd = 'python3', '-c', 'import time; [(print(i), time.sleep(1)) for i in range(5)]'
p = subprocess.Popen(cmd, stdout=subprocess.PIPE)
os.set_blocking(p.stdout.fileno(), False)
start = time.time()
while True:
    # first iteration always produces empty byte string in non-blocking mode
    for i in range(2):    
        line = p.stdout.readline()
        print(i, line)
        time.sleep(0.5)
    if time.time() > start + 5:
        break
p.terminate()

Dies gibt aus:

1 b''
2 b'0\n'
1 b''
2 b'1\n'
1 b''
2 b'2\n'
1 b''
2 b'3\n'
1 b''
2 b'4\n'

Mit os.set_blockingkommentiert ist es:

0 b'0\n'
1 b'1\n'
0 b'2\n'
1 b'3\n'
0 b'4\n'
1 b''
saaj
quelle
-2

Hier ist ein Modul, das nicht blockierende Lese- und Hintergrundschreibvorgänge in Python unterstützt:

https://pypi.python.org/pypi/python-nonblock

Bietet eine Funktion,

nonblock_read, das Daten aus dem Stream liest, falls verfügbar, andernfalls eine leere Zeichenfolge zurückgibt (oder None, wenn der Stream auf der anderen Seite geschlossen ist und alle möglichen Daten gelesen wurden).

Sie können auch das Modul python-subprocess2 in Betracht ziehen.

https://pypi.python.org/pypi/python-subprocess2

Dies fügt dem Unterprozessmodul hinzu. Für das von "subprocess.Popen" zurückgegebene Objekt wird eine zusätzliche Methode hinzugefügt, runInBackground. Dadurch wird ein Thread gestartet und ein Objekt zurückgegeben, das automatisch ausgefüllt wird, wenn Inhalte in stdout / stderr geschrieben werden, ohne den Hauptthread zu blockieren.

Genießen!

Tim Savannah
quelle
Ich würde dieses Nonblock- Modul gerne ausprobieren, bin aber bei einigen Linux-Verfahren relativ neu. Wie genau installiere ich diese Routinen? Ich verwende Raspbian Jessie, eine Variante von Debian Linux für den Raspberry Pi. Ich habe 'sudo apt-get install nonblock' und python-nonblock ausprobiert und beide haben einen Fehler ausgegeben - nicht gefunden. Ich habe die Zip-Datei von dieser Site pypi.python.org/pypi/python-nonblock heruntergeladen , weiß aber nicht, was ich damit machen soll. Vielen Dank .... RDK
RDK