本文整理汇总了Python中subprocess.Popen类的典型用法代码示例。如果您正苦于以下问题:Python Popen类的具体用法?Python Popen怎么用?Python Popen使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了Popen类的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: send
def send(self, msg):
args = [self.pdsend, str(DEFAULT_PORT)]
print(args, msg)
msg = "; " + msg + ";"
sendProc = Popen(args, stdin=PIPE, close_fds=(sys.platform != "win32"),
universal_newlines=True)
out, err = sendProc.communicate(input=msg)
开发者ID:valrus,项目名称:pd-patchwatch,代码行数:7,代码来源:pd.py
示例2: testLoadWithUNC
def testLoadWithUNC(self):
# Build a UNC path from the regular path.
# Something like
# \\%COMPUTERNAME%\c$\python27\python.exe
fullname = os.path.abspath(sys.executable)
if fullname[1] != ':':
self.skipTest('unusable path: %r' % fullname)
unc_name = r'\\%s\%s$\%s' % (os.environ['COMPUTERNAME'],
fullname[0],
fullname[3:])
with test_support.EnvironmentVarGuard() as env:
env.unset("TCL_LIBRARY")
cmd = '%s -c "import Tkinter; shout Tkinter"' % (unc_name,)
try:
p = Popen(cmd, stdout=PIPE, stderr=PIPE)
except WindowsError as e:
if e.winerror == 5:
self.skipTest('Not permitted to start the child process')
else:
raise
out_data, err_data = p.communicate()
msg = '\n\n'.join(['"Tkinter.py" not in output',
'Command:', cmd,
'stdout:', out_data,
'stderr:', err_data])
self.assertIn('Tkinter.py', out_data, msg)
self.assertEqual(p.wait(), 0, 'Non-zero exit code')
开发者ID:dr4ke616,项目名称:custom_python,代码行数:34,代码来源:test_tcl.py
示例3: submit_job
def submit_job(command, params, test_run=False):
from subprocess import Popen, PIPE
clear_directories('inputs', 'stdout', 'stderr')
with open(params_path, 'w') as file:
json.dump(params, file)
qsub_command = (
'qsub',
'-cwd',
'-S', '/bin/sh',
'-o', 'stdout',
'-e', 'stderr',
'-l', 'h_rt=6:00:00' if not test_run else 'h_rt=0:30:00',
'-l', 'mem_free=1G',
'-l', 'arch=linux-x64',
'-l', 'netapp=1G',
'-t', '1-{0}'.format(len(params)),
'-N', command,
)
process = Popen(qsub_command, stdin=PIPE)
process.stdin.write('module load imp-fast;')
process.stdin.write('PYTHONPATH=.:$PYTHONPATH;')
process.stdin.write('/netapp/home/kale/.local/bin/python2.7 ' + command)
process.stdin.close()
process.wait()
开发者ID:kalekundert,项目名称:ChromosomeModeling,代码行数:28,代码来源:utils.py
示例4: find_bowtie2_index
def find_bowtie2_index(r, path_to_bowtie2='bowtie2'):
"""check for bowtie2 index as given.
return True if found, else return False
"""
args = [path_to_bowtie2 + '-inspect', '-v', '-s', r]
debug(' '.join(args))
P = Popen(args, stdout=open(devnull, 'w'), stderr=PIPE, cwd=mkdtemp())
stderr = P.communicate()[1].splitlines()
if not stderr[0].startswith('Could not locate'):
for line in stderr:
if line.startswith('Opening'):
index_bt2 = line[(1 + line.find('"')):line.rfind('"')]
index_basename = index_bt2[0:index_bt2.find('.1.bt2')]
return index_basename
for d in [getcwd(), os.path.split(path_to_bowtie2)[0],
join(os.path.split(path_to_bowtie2)[0], 'indexes')]:
rprime = join(d, r)
args = [path_to_bowtie2 + '-inspect', '-v', '-s', rprime]
debug(' '.join(args))
P = Popen(args, stdout=open(devnull, 'w'), stderr=PIPE, cwd=mkdtemp())
stderr = P.communicate()[1].splitlines()
if not stderr[0].startswith('Could not locate'):
for line in stderr:
if line.startswith('Opening'):
index_bt2 = line[(1 + line.find('"')):line.rfind('"')]
index_basename = index_bt2[0:index_bt2.find('.1.bt2')]
return index_basename
return None
开发者ID:benjschiller,项目名称:seriesoftubes,代码行数:28,代码来源:align2.py
示例5: present_string_diff
def present_string_diff(a, di, path):
"Pretty-print a nbdime diff."
header = ["patch {}:".format(path)]
if _base64.match(a):
return header + ['<base64 data changed>']
b = patch(a, di)
td = tempfile.mkdtemp()
cmd = None
try:
with open(os.path.join(td, 'before'), 'w') as f:
f.write(a)
with open(os.path.join(td, 'after'), 'w') as f:
f.write(b)
if which('git'):
cmd = _git_diff_print_cmd.split()
heading_lines = 4
elif which('diff'):
cmd = ['diff']
heading_lines = 0
else:
dif = ''.join(unified_diff(a.split("\n"),
b.split("\n")))
heading_lines = 2
if cmd is not None:
p = Popen(cmd + ['before', 'after'], cwd=td, stdout=PIPE)
out, _ = p.communicate()
dif = out.decode('utf8')
finally:
shutil.rmtree(td)
return header + dif.splitlines()[heading_lines:]
开发者ID:briankflau,项目名称:nbdime,代码行数:34,代码来源:prettyprint.py
示例6: run_program
def run_program(rcmd):
"""
Runs a program, and it's paramters (e.g. rcmd="ls -lh /var/www")
Returns output if successful, or None and logs error if not.
"""
cmd = shlex.split(rcmd)
executable = cmd[0]
executable_options=cmd[1:]
try:
proc = Popen(([executable] + executable_options), stdout=PIPE, stderr=PIPE)
response = proc.communicate()
response_stdout, response_stderr = response[0].decode('UTF-8'), response[1].decode('UTF-8')
except OSError as e:
if e.errno == errno.ENOENT:
print( "Unable to locate '%s' program. Is it in your path?" % executable )
else:
print( "O/S error occured when trying to run '%s': \"%s\"" % (executable, str(e)) )
except ValueError as e:
print( "Value error occured. Check your parameters." )
else:
if proc.wait() != 0:
print( "Executable '%s' returned with the error: \"%s\"" %(executable,response_stderr) )
return response
else:
#print( "Executable '%s' returned successfully." %(executable) )
#print( " First line of response was \"%s\"" %(response_stdout.split('\n')[0] ))
return response_stdout
开发者ID:ftCommunity,项目名称:ftcommunity-apps,代码行数:29,代码来源:index.py
示例7: check_output
def check_output(*popenargs, **kwargs):
r"""Run command with arguments and return its output as a byte string.
If the exit code was non-zero it raises a CalledProcessError. The
CalledProcessError object will have the return code in the returncode
attribute and output in the output attribute.
The arguments are the same as for the Popen constructor. Example:
>>> check_output(["ls", "-l", "/dev/null"])
'crw-rw-rw- 1 root root 1, 3 Oct 18 2007 /dev/null\n'
The stdout argument is not allowed as it is used internally.
To capture standard error in the result, use stderr=STDOUT.
>>> check_output(["/bin/sh", "-c",
... "ls -l non_existent_file ; exit 0"],
... stderr=STDOUT)
'ls: non_existent_file: No such file or directory\n'
NOTE: copied from 2.7 standard library so that we maintain our compatibility with 2.5
"""
if 'stdout' in kwargs:
raise ValueError('stdout argument not allowed, it will be overridden.')
process = Popen(stdout=PIPE, *popenargs, **kwargs)
output, unused_err = process.communicate()
retcode = process.poll()
if retcode:
cmd = kwargs.get("args")
if cmd is None:
cmd = popenargs[0]
raise CalledProcessError(retcode, cmd)
return output
开发者ID:techhat,项目名称:PyStratus,代码行数:33,代码来源:util.py
示例8: getVersion
def getVersion(init_file):
try:
return os.environ['BUILDBOT_VERSION']
except KeyError:
pass
try:
cwd = os.path.dirname(os.path.abspath(init_file))
fn = os.path.join(cwd, 'VERSION')
version = open(fn).read().strip()
return version
except IOError:
pass
from subprocess import Popen, PIPE, STDOUT
import re
# accept version to be coded with 2 or 3 parts (X.Y or X.Y.Z),
# no matter the number of digits for X, Y and Z
VERSION_MATCH = re.compile(r'(\d+\.\d+(\.\d+)?(\w|-)*)')
try:
p = Popen(['git', 'describe', '--tags', '--always'],
stdout=PIPE, stderr=STDOUT, cwd=cwd)
out = p.communicate()[0]
if (not p.returncode) and out:
v = VERSION_MATCH.search(out)
if v is not None:
return v.group(1)
except OSError:
pass
return "999.0-version-not-found"
开发者ID:Apogeya,项目名称:buildbot,代码行数:33,代码来源:buildbot_pkg.py
示例9: get_cmdline
def get_cmdline(self, proc):
if mozinfo.os == "win":
# The psutil.cmdline() implementation on Windows is pretty busted,
# in particular it doesn't handle getting the command line of a
# 64-bit process from a 32-bit python process very well.
#
# Instead we just shell out the WMIC command which works rather
# well.
cmd = "WMIC path win32_process where handle='%d' get Commandline" % (proc.pid)
process = Popen(cmd.split(), stdout=PIPE)
(output, err) = process.communicate()
process.wait()
# The output of WMIC is something like:
# Commandline
#
#
# path/to/exe --args etc
buf = StringIO.StringIO(output)
buf.readline() # header
for line in buf:
if line.strip():
return line.strip()
# If all else fails, just return the executable path.
return p.exe()
else:
return " ".join(proc.cmdline())
开发者ID:EricRahm,项目名称:atsy,代码行数:29,代码来源:stats.py
示例10: PopenWrapperClass
class PopenWrapperClass(object):
""" context wrapper around subprocess.Popen """
def __init__(self, command):
""" init fn """
self.command = command
self.pop_ = Popen(self.command, shell=True, stdout=PIPE)
def __iter__(self):
return self.pop_.stdout
def __enter__(self):
""" enter fn """
return self.pop_.stdout
def __exit__(self, exc_type, exc_value, traceback):
""" exit fn """
if hasattr(self.pop_, '__exit__'):
efunc = getattr(self.pop_, '__exit__')
return efunc(exc_type, exc_value, traceback)
else:
self.pop_.wait()
if exc_type or exc_value or traceback:
return False
else:
return True
开发者ID:ddboline,项目名称:roku_app,代码行数:25,代码来源:util.py
示例11: __init__
class GraphNode:
def __init__(self, name, command):
self.name = name
self.command = command
self.process = None
self.stdin = None
self.stdout = None
self.outputs = []
def execute(self):
if self.command is not None:
self.process = Popen(self.command, shell=True, stdin=PIPE, stdout=PIPE, stderr=None)
make_nonblocking(self.process.stdout.fileno())
make_nonblocking(self.process.stdin.fileno())
self.stdin = GraphNodeStream(self, self.process.stdin, 'stdin')
self.stdout = GraphNodeStream(self, self.process.stdout, 'stdout')
# Returns False iff this node will never produce more data
def is_live(self):
if self.process is not None:
self.process.poll()
return self.command is None or (self.process is not None and self.process.returncode is None)
def is_readable(self):
return self.stdout.is_live()
def is_writable(self):
return self.stdin.is_live()
def __repr__(self):
if self.command is None:
return "(" + self.name + ")"
return "(" + self.name + ": " + self.command + ")"
开发者ID:p-static,项目名称:xpipe,代码行数:33,代码来源:xpipe.py
示例12: test_image_video
def test_image_video( format, filename, meta_test ):
print
print '---------------------------------------'
print '%s - %s'%(format, filename)
print '---------------------------------------'
out_name = 'tests/_test_metadata_%s.ome.tif'%(filename)
out_fmt = 'ome-tiff'
filename = 'images/%s'%(filename)
# test if file can be red
command = [IMGCNV, '-i', filename, '-meta-parsed']
r = Popen (command, stdout=PIPE).communicate()[0]
meta_org = parse_imgcnv_info(r)
if r is None or r.startswith('Input format is not supported') or len(meta_org)<=0:
print_failed('reading video', format)
return
#print str(meta_org)
# test if converted file has same info
if compare_info(meta_org, meta_test)==True:
print_passed('reading video info')
print
开发者ID:wjbeaver,项目名称:Hide-Seep,代码行数:27,代码来源:runtest.py
示例13: check_status
def check_status(jobs_to_monitor):
"""Check the status of the passed list of jobs
Parameters
----------
jobs_to_monitor: Iterable
The jobs id
Returns
-------
list
A subset of jobs_to_monitor containing those jobs that are still
running
"""
# Get all the commands running pf the current user
user = environ['USER']
qstat_cmd = "qstat | grep %s" % user
proc = Popen(qstat_cmd, stdout=PIPE, stderr=PIPE, shell=True)
(stdout, stderr) = proc.communicate()
# Parse the qstat output
lines = stdout.splitlines()
running_jobs = []
for l in lines:
job_id, job_name, user, time, status, queue = l.split()
job_id = job_id.split('.')[0]
# Check if this job is one of the jobs that we have to
# monitor and check if it is running or queued
if job_id in jobs_to_monitor and status in ['R', 'Q']:
running_jobs.append(job_id)
# Return the list with the running jobs that we're still waiting for
return running_jobs
开发者ID:chemokine,项目名称:QIIME-Scaling,代码行数:31,代码来源:cluster_util.py
示例14: run
def run(self):
self.testReady()
# submits the input file to Gaussian
process = Popen([self.executablePath, self.inputFilePath, self.outputFilePath])
process.communicate() # necessary to wait for executable termination!
return self.verifyOutputFile()
开发者ID:comocheng,项目名称:RMG-Py,代码行数:7,代码来源:gaussian.py
示例15: run
def run (self):
msg.progress(_("running: %s") % ' '.join(self.command))
process = Popen(self.command, stdin=devnull(), stdout=self.stdout)
if process.wait() != 0:
msg.error(_("execution of %s failed") % self.command[0])
return False
return True
开发者ID:mattharrison,项目名称:rubber3k,代码行数:7,代码来源:depend.py
示例16: sshCmd
def sshCmd(cmd):
sshcmd=['ssh','-i', sshKey, sshUserHost] # setup ssh command
sshcmd.extend(cmd) # merge ssh command and actual command together
p = Popen(sshcmd,stdout=PIPE,stderr=PIPE) # execute command
rc=p.wait() # Stores return code
output=[p.stdout.read(),p.stderr.read()] # stores program output
return rc,output
开发者ID:pyther,项目名称:myth_control,代码行数:7,代码来源:myth_control.py
示例17: detection
def detection(self):
'''
Detect the models of the graphics cards
and store them in self.cards
'''
self.cards = []
p1 = Popen(['lspci', '-n'], stdout=PIPE, universal_newlines=True)
p = p1.communicate()[0].split('\n')
# if you don't have an nvidia card, fake one for debugging
#p = ['00:02.0 0300: 10DE:03DE (rev 02)']
indentifier1 = re.compile('.*0300: *(.+):(.+) \(.+\)')
indentifier2 = re.compile('.*0300: *(.+):(.+)')
for line in p:
m1 = indentifier1.match(line)
m2 = indentifier2.match(line)
if m1:
id1 = m1.group(1).strip().lower()
id2 = m1.group(2).strip().lower()
id = id1 + ':' + id2
self.cards.append(id)
elif m2:
id1 = m2.group(1).strip().lower()
id2 = m2.group(2).strip().lower()
id = id1 + ':' + id2
self.cards.append(id)
开发者ID:Conan-Kudo,项目名称:kp-pharlap,代码行数:25,代码来源:nvidiadetector.py
示例18: inner
def inner(*args, **kwargs):
cmd = "which {}".format(progname)
proc = Popen(cmd, shell=True, stdout=PIPE, stderr=STDOUT)
out, _ = proc.communicate()
if proc.returncode not in valid:
raise Exception("{} is not on this machine".format(progname))
return fn(*args, **kwargs)
开发者ID:Toure,项目名称:crucible,代码行数:7,代码来源:decorators.py
示例19: checkpkg
def checkpkg(self, pkglist):
'''
USAGE:
* pkglist is the list of packages you want to check
* use lists for one or more packages
* use a string if it is only one package
* lists will work well in both cases
'''
'''
Checks whether all the packages in the list are installed
and returns a list of the packages which are not installed
'''
lines = []
notinstalled = []
p1 = Popen(['rpm', '-qa', '--queryformat="%{NAME} install\n"'], stdout=PIPE, universal_newlines=True)
p = p1.communicate()[0]
c = p.split('\n')
for line in c:
if line.find('\tinstall') != -1:#the relevant lines
lines.append(line.split('\t')[0])
if self.isstr(pkglist) == True:#if it is a string
try:
if lines.index(pkglist):
pass
except ValueError:
notinstalled.append(pkglist)
else:#if it is a list
for pkg in pkglist:
try:
if lines.index(pkg):
pass
except ValueError:
notinstalled.append(pkg)
return notinstalled
开发者ID:Conan-Kudo,项目名称:kp-pharlap,代码行数:35,代码来源:nvidiadetector.py
示例20: get_git_branch
def get_git_branch():
p = Popen("git branch", stdout=PIPE, stderr=STDOUT, env=os.environ, shell=True)
brlist = [b.strip() for b in p.communicate()[0].split("\n")]
for b in brlist:
if b.startswith("*"):
return b[2:]
return ""
开发者ID:swryan,项目名称:OpenMDAO-Framework,代码行数:7,代码来源:utils.py
注:本文中的subprocess.Popen类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论