I have created a class to communicate with another processes via pipes.
Class creates separate threads that read/write to pipes and use async queues to communicate with your thread.
It;s a proven solution that I use in my project
import time
import subprocess
import queue
import threading
TIMEOUT_POLLINGINTERVAL = 0.5
class ShellCall():
def __init__(self):
self._popen = None
""" :type: subprocess.Popen """
self._stdOutQueue = None
""" :type: queue.Queue """
self._stdErrQueue = None
""" :type: queue.Queue """
self._stdOut = []
self._stdErr = []
def __del__(self):
if self._popen and self._popen.poll() is None:
self._popen.kill()
def call(self, command, shell=False):
"""
Execute a shell command
:param command: command to be executed
:type command: str | list[str]
:param shell: If shell is True, the specified command will be executed through the shell
:type shell: bool
:rtype: None
"""
if shell:
command = command.encode('utf-8')
else:
command = [item.encode('utf-8') for item in command]
self._popen = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=shell, bufsize=0)
self._stdOutQueue = self._getAsyncReadQueue(self._popen.stdout)
self._stdErrQueue = self._getAsyncReadQueue(self._popen.stderr)
def _getAsyncReadQueue(self, sourcePipe):
"""
Create a thread to read from pipe in asynchronous mode, get queue to receive data from pipe
:param sourcePipe: Pipe to read from
:type sourcePipe: pipe
:return: Queue to receive read data
:rtype: queue.Queue
"""
newQueue = queue.Queue()
newThread = threading.Thread(target=self._enqueueOutput, args=(sourcePipe, newQueue))
newThread.daemon = True # thread dies with the program
newThread.start()
return newQueue
@staticmethod
def _enqueueOutput(sourcePipe, outputQueue):
"""
Read from pipe and write to the queue
:param sourcePipe: Pipe to read from
:type sourcePipe: pipe
:param outputQueue: Queue to write to
:type outputQueue: queue.Queue
"""
for line in iter(sourcePipe.readline, b''):
outputQueue.put(line)
def waitNotNoneReturnCode(self, timeout, *, checkCallback=None):
"""
Wait until any return code
:param timeout: Timeout for executed command (sec). If timeout expired - ShellException raised
:type timeout: float
:param checkCallback: Any callable that will be used to check is shell call finished
:type checkCallback: callable
:rtype: None
"""
self._wait(timeout, notNoneReturnCode=True, checkCallback=checkCallback)
def waitNoErrorReturnCode(self, timeout, *, checkCallback=None):
"""
Wait until success return code '0'. Otherwise raise ShellException
:param timeout: Timeout for executed command (sec). If timeout expired - ShellException raised
:type timeout: float
:param checkCallback: Any callable that will be used to check is shell call finished
:type checkCallback: callable
:rtype: None
"""
self._wait(timeout, notNoneReturnCode=True, noErrorReturnCode=True, checkCallback=checkCallback)
def waitNoStdErr(self, timeout, *, checkCallback=None):
"""
Wait until success return code '0' and empty stderr. Otherwise raise ShellException
:param timeout: Timeout for executed command (sec). If timeout expired - ShellException raised
:type timeout: float
:param checkCallback: Any callable that will be used to check is shell call finished
:type checkCallback: callable
:rtype: None
"""
self._wait(timeout, notNoneReturnCode=True, noErrorReturnCode=True, noStdErr=True, checkCallback=checkCallback)
def waitStdOut(self, timeout, *, checkCallback=None):
"""
Wait until success return code '0', empty stderr and not empty stdout. Otherwise raise ShellException
:param timeout: Timeout for executed command (sec). If timeout expired - ShellException raised
:type timeout: float
:param checkCallback: Any callable that will be used to check is shell call finished
:type checkCallback: callable
:rtype: None
"""
self._wait(timeout, notNoneReturnCode=True, noErrorReturnCode=True,
noStdErr=True, stdOut=True, checkCallback=checkCallback)
def _wait(self, timeout, *, pollingTime=TIMEOUT_POLLINGINTERVAL,
notNoneReturnCode=False, noErrorReturnCode=False, noStdErr=False, stdOut=False, checkCallback=None):
"""
Raise ShellException if conditions not satisfied (see :func:`checkCallResults`).
Raise ShellException if conditions not satisfied too long.
:param timeout: Timeout for executed command (sec). If timeout expired - ShellException raised
:type timeout: float
:param pollingTime: Time interval length to check result of command execution
:type pollingTime: float
:rtype: None
"""
startTime = time.time()
while True:
if self._checkCallResults(notNoneReturnCode=notNoneReturnCode, noErrorReturnCode=noErrorReturnCode,
noStdErr=noStdErr, stdOut=stdOut, checkCallback=checkCallback):
return
# exception due to timeout
if time.time() - startTime > timeout:
raise ShellException('Shell call not finished too long', self)
time.sleep(pollingTime)
def _checkCallResults(self, notNoneReturnCode=False, noErrorReturnCode=False,
noStdErr=False, stdOut=False, checkCallback=None):
"""
Raise ShellException if noErrorReturnCode=True and shell call return not 0 return call
Raise ShellException if noStdErr=True and shell call print anything to stderr
:param notNoneReturnCode: return True only if shell call return any return call
:type notNoneReturnCode: bool
:param noErrorReturnCode: return True only if shell call return 0 return code
:type noErrorReturnCode: bool
:param noStdErr: return True only if shell call print nothing to stderr
:type noStdErr: bool
:param stdOut: return True only if shell call print anything to stdout
:type stdOut: bool
:param checkCallback: Any callable that will be used to check is shell call finished,
positional arguments, keyword arguments
:type checkCallback: callable, args, kwargs
:return: True if conditions are satisfied
:rtype: bool
"""
# exceptions
if noErrorReturnCode:
if self.getReturnCode() is not None and self.getReturnCode() > 0:
raise ShellException('Shell call finished with error return code', self)
if noStdErr:
if len(self.getStdErr()) > 0:
raise ShellException('Shell call have non-empty stderr', self)
# break loop
notNoneReturnCodeCondition = (self.getReturnCode() is not None) if notNoneReturnCode else True
noErrorReturnCodeCondition = (self.getReturnCode() == 0) if noErrorReturnCode else True
notStdErrCondition = (len(self.getStdErr()) == 0) if noStdErr else True
stdOutCondition = (len(self.getStdOut()) > 0) if stdOut else True
callbackCondition = checkCallback() if checkCallback else True
if notNoneReturnCodeCondition and noErrorReturnCodeCondition and \
notStdErrCondition and stdOutCondition and callbackCondition:
return True
else:
return False
def getReturnCode(self):
"""
Get return code of the process
:return: return code of the child process or None if process is not terminated yet
:rtype: int|None
"""
return self._popen.poll()
def getStdOut(self):
"""
Get list with stdout lines
:rtype: list[str]
"""
self._stdOut += self._readAllQueue(self._stdOutQueue)
return self._stdOut
def getStdErr(self):
"""
Get list with stderr lines
:rtype: list[str]
"""
self._stdErr += self._readAllQueue(self._stdErrQueue)
return self._stdErr
@staticmethod
def _readAllQueue(sourceQueue):
lines = []
try:
while True:
line = sourceQueue.get_nowait() # or q.get(timeout=.1)
line = line.decode('utf-8').rstrip()
lines.append(line)
except queue.Empty:
return lines
def __repr__(self):
stdOut = str.join(' ', self.getStdOut())
stdOut = (stdOut[:1000] + '...') if len(stdOut) > 1000 else stdOut
stdErr = str.join(' ', self.getStdErr())
stdErr = (stdErr[:1000] + '...') if len(stdErr) > 1000 else stdErr
return '<ShellCall(command={}, ReturnCode={}, stdout="{}", stderr="{}")>'. \
format(self._popen.args, self.getReturnCode(), stdOut, stdErr)
class ShellException(Exception):
def __init__(self, description, shellCall):
"""
:param description: test description of the error
:type description: str
:param shellCall: shell call object used to execute a command
:type shellCall: ShellCall
:rtype: None
"""
super(Exception, self).__init__(description, shellCall)
def getShellCall(self):
"""
Get shell call object used to execute a command
:rtype: ShellCall
"""
description, shellCall = self.args
return shellCall
