as you can see, this code takes a list of commands and runs them in a stdout to stdin pipeline. it returns the read side of the pipe from the last process in the pipeline.
from os import write
from subprocess import PIPE,Popen
from sys import argv,stderr
from time import sleep
def cmdsout(cmds,**opts):
"""Run a command (list/tuple of list/tuple of str/bytes/bytearray)
returning its last output as a big str/bytes/bytearray buffer of newline separated lines."""
# for POSIX/BSD/Unix/Linux systems
# tested on Ubuntu Linux
# each command can be either a list or tuple
# each command arg must be the same type within a command, one of str or bytes or bytearray
# but can be different between commands
# the command pipeline can be either a list or tuple
if not isinstance(cmds,(list,tuple)):
raise TypeError('argument 1 (cmds) must be a list or tuple (of list/tuple of str/bytes/bytearray).')
if not all(isinstance(cmd,(list,tuple))for cmd in cmds):
raise TypeError('argument 1 (cmds) must be a list or tuple of list or tuple (of str/bytes/bytearray).')
if not all(all(isinstance(arg,(str,bytes,bytearray))for arg in cmd)for cmd in cmds):
raise TypeError('argument 1 (cmds) must be a list or tuple of list or tuple of str or bytes or bytearray.')
for num in range(len(cmds)):
if not all(isinstance(arg,type(cmds[num][0]))for arg in cmds[num]):
print('cmd =',repr(cmd),flush=1)
raise TypeError(f'argument 1 (cmds) cmd #{num} has not all args the same type of str/bytes/bytearray.')
procs = []
for cmd in cmds[:-1]:
procs.append(Popen(cmd,stdout=PIPE,universal_newlines=isinstance(cmd[0],str),**opts))
opts['stdin']=procs[-1].stdout # connect next stdin to previous stdout
with Popen(cmds[-1],stdout=PIPE,universal_newlines=isinstance(cmds[-1][0],str),**opts) as lastproc:
return (lastproc.stdout)
version = '0.0.0'
def main(cmdpath,args):
slow = 1/32
# test the cmdsout() function with a little command pipeline that lists
# files with ls, filters the ls output for python files, and sorts the
# output by date/time.
# each command is a list or tuple of str or bytes or bytearray
cmd1=('ls','-l','--full-time')+tuple(args)
cmd2=[b'egrep',b'\.py$']
cmd3=[b'sort',b'-k6']
# each command pipeline is a list or tuple of commands
cmds=(cmd1,cmd2,cmd3)
# the command pipeline is passed to cmdsout in the only argument
# the return value is the stdout file
out=cmdsout(cmds)
print('',file=stderr)
for line in out:
line=line.rstrip()
sleep(slow)
write(1,line)
write(1,b'\n')
print('',file=stderr)
print(f'got a {type(out).__name__} of {type(line).__name__}',file=stderr)
return 0
def help():
print('See a time sorted list of Python files.')
return
def version():
print(version)
return
if __name__ == '__main__':
cmdpath = argv.pop(0)
if argv and argv[0][:2]=='--':
if '--help' in argv:
help()
if '--version' in argv:
version()
exit(1)
try:
result=main(cmdpath,argv)
except BrokenPipeError:
exit(141)
except KeyboardInterrupt:
print(flush=True)
exit(98)
if isinstance(result,(bytes,bytearray)):
result=''.join(chr(x)for x in result)
if isinstance(result,str):
print(result,file=stderr,flush=True)
result=1
if result is None or result is True:
result=0
elif result is False:
result=1
exit(int(float(result)))