本文整理汇总了Python中psutil.Popen类的典型用法代码示例。如果您正苦于以下问题：Python Popen类的具体用法？Python Popen怎么用？Python Popen使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了Popen类的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: addFile
def addFile(self, sourcefile):
if self.type in ['RAR', 'RAR5']:
raise NotImplementedError
process = Popen('7z a -y "' + self.filepath + '" "' + sourcefile + '"',
stdout=PIPE, stderr=STDOUT, stdin=PIPE, shell=True)
process.communicate()
if process.returncode != 0:
raise OSError('Failed to add the file.')
开发者ID:ciromattia，项目名称:kcc，代码行数:8，代码来源:comicarchive.py
示例2: test_get_info
def test_get_info(self):
worker = Popen(["python -c 'import time;time.sleep(5)'"], shell=True)
try:
info = get_info(worker)
finally:
worker.terminate()
self.assertTrue(isinstance(info["pid"], int))
self.assertEqual(info["nice"], 0)
开发者ID:nightshade427，项目名称:circus，代码行数:9，代码来源:test_util.py
示例3: test_get_info
def test_get_info(self):
worker = Popen(['top'], shell=True)
try:
info = get_info(worker)
finally:
worker.terminate()
self.assertTrue(isinstance(info['pid'], int))
self.assertEqual(info['nice'], 0)
开发者ID:msabramo，项目名称:circus，代码行数:10，代码来源:test_util.py
示例4: popen_wrapper
def popen_wrapper(args, universal_newlines=True):
p = Popen(args, shell=False, stdout=PIPE, stderr=PIPE,
close_fds=os.name != 'nt', universal_newlines=universal_newlines)
output, errors = p.communicate()
return (
output,
errors,
p.returncode
)
开发者ID:python-bot，项目名称:python-bot，代码行数:10，代码来源:subprocess.py
示例5: extractMetadata
def extractMetadata(self):
process = Popen('7z x -y -so "' + self.filepath + '" ComicInfo.xml',
stdout=PIPE, stderr=STDOUT, stdin=PIPE, shell=True)
xml = process.communicate()
if process.returncode != 0:
raise OSError('Failed to extract archive.')
try:
return parseString(xml[0])
except ExpatError:
return None
开发者ID:ciromattia，项目名称:kcc，代码行数:10，代码来源:comicarchive.py
示例6: system
def system(cmd, timeout = 60):
global proc
logger().info(u"Executing command '{0}'".format(cmd))
proc = Popen(cmd, close_fds=True, shell=True, stdout=PIPE)
logger().info(u"PID is {0} for command '{1}'".format(proc.pid, cmd))
for line in proc.stdout:
logger().info(u"Process output: {0}".format(line.decode('latin1')))
## Kill process after timeout seconds.
_timer = Timer(timeout, terminate_process, [proc])
_timer.start()
proc.wait(timeout=3) ## Wait for process to complete, increase timeout parameter if default 3 seconds is not enough
_timer.cancel()
return proc
开发者ID:olefriis，项目名称:simplepvr，代码行数:13，代码来源:hdhomerun.py
示例7: test_get_info
def test_get_info(self):
worker = Popen(["python", "-c", SLEEP % 5])
try:
info = get_info(worker)
finally:
worker.terminate()
self.assertTrue(isinstance(info['pid'], int))
if IS_WINDOWS:
self.assertEqual(info['nice'], psutil.NORMAL_PRIORITY_CLASS)
else:
self.assertEqual(info['nice'], 0)
开发者ID:BrainBot，项目名称:circus，代码行数:13，代码来源:test_util.py
示例8: raw_cmd
def raw_cmd(arguments, shell=False, **kwargs):
"""
Launch a subprocess.
This function ensure that:
* subprocess arguments will be converted to a string if `shell` is True
* subprocess.args is set to the arguments of the subprocess
"""
arguments_list = to_args_list(arguments)
process = Popen(to_args_string(arguments_list) if shell else arguments_list, shell=shell, **kwargs)
if not hasattr(process, "args"):
process.args = arguments_list
return process
开发者ID:tonysepia，项目名称:pytoolbox，代码行数:14，代码来源:subprocess.py
示例9: __init__
def __init__(self):
# Need to find GNU time. If /usr/bin/time is not GNU time
# then PUQ expects it to be in the path and called 'gtime'
tstr = 'gtime'
try:
ver = Popen("/usr/bin/time --version", shell=True, stderr=PIPE).stderr.read()
if ver.startswith(b"GNU"):
tstr = '/usr/bin/time'
except:
pass
self.timestr = tstr + " -f \"HDF5:{'name':'time','value':%e,'desc':''}:5FDH\""
self.run_num = 0
开发者ID:c-PRIMED，项目名称:puq，代码行数:14，代码来源:hosts.py
示例10: __init__
def __init__(self, filepath):
self.filepath = filepath
self.type = None
if not os.path.isfile(self.filepath):
raise OSError('File not found.')
process = Popen('7z l -y -p1 "' + self.filepath + '"', stderr=STDOUT, stdout=PIPE, stdin=PIPE, shell=True)
for line in process.stdout:
if b'Type =' in line:
self.type = line.rstrip().decode().split(' = ')[1].upper()
break
process.communicate()
if process.returncode != 0:
raise OSError('Archive is corrupted or encrypted.')
elif self.type not in ['7Z', 'RAR', 'RAR5', 'ZIP']:
raise OSError('Unsupported archive format.')
开发者ID:ciromattia，项目名称:kcc，代码行数:15，代码来源:comicarchive.py
示例11: make
def make(base_dir, timeout=5):
"""compile source code.
return (ok?, msg)"""
cmd = ['clang++', '-std=c++11', 'main.cpp']
p = Popen(cmd, stderr=PIPE, cwd=base_dir)
try:
p.wait(timeout)
except TimeoutExpired:
return False, 'compilation take too much time.'
else:
if p.returncode == 0:
return True, ''
else:
return False, p.communicate()[1]
开发者ID:jhonprice，项目名称:yaoj-django，代码行数:15，代码来源:tasks.py
示例12: extract
def extract(self, targetdir):
if not os.path.isdir(targetdir):
raise OSError('Target directory don\'t exist.')
process = Popen('7z x -y -xr!__MACOSX -xr!.DS_Store -xr!thumbs.db -xr!Thumbs.db -o"' + targetdir + '" "' +
self.filepath + '"', stdout=PIPE, stderr=STDOUT, stdin=PIPE, shell=True)
process.communicate()
if process.returncode != 0:
raise OSError('Failed to extract archive.')
tdir = os.listdir(targetdir)
if 'ComicInfo.xml' in tdir:
tdir.remove('ComicInfo.xml')
if len(tdir) == 1 and os.path.isdir(os.path.join(targetdir, tdir[0])):
for f in os.listdir(os.path.join(targetdir, tdir[0])):
move(os.path.join(targetdir, tdir[0], f), targetdir)
os.rmdir(os.path.join(targetdir, tdir[0]))
return targetdir
开发者ID:ciromattia，项目名称:kcc，代码行数:16，代码来源:comicarchive.py
示例13: run_command
def run_command(cmd, data, location, chw):
cwd = os.getcwd()
if location is not None and chw is True:
cwd = location
elif location is not None and chw is False:
cmd = '{0} {1}'.format(cmd, location)
r = Popen(shlex.split(cmd), stdout=PIPE, stdin=PIPE, stderr=PIPE, cwd=cwd)
if data is None:
output = r.communicate()[0].decode('utf-8')
else:
output = r.communicate(input=data)[0]
return output
开发者ID:ttsvetanov，项目名称:polyaxon，代码行数:16，代码来源:git.py
示例14: __init__
def __init__(self, wid, cmd, working_dir, shell, uid=None, gid=None, env=None):
self.wid = wid
self.working_dir = working_dir
self.shell = shell
self.env = env
self.cmd = cmd.replace("$WID", str(self.wid))
self.uid = to_uid(uid)
self.gid = to_gid(gid)
def preexec_fn():
os.setsid()
if self.gid:
try:
os.setgid(self.gid)
except OverflowError:
if not ctypes:
raise
# versions of python < 2.6.2 don't manage unsigned int for
# groups like on osx or fedora
os.setgid(-ctypes.c_int(-self.gid).value)
if self.uid:
os.setuid(self.uid)
self._worker = Popen(
self.cmd.split(),
cwd=self.working_dir,
shell=self.shell,
preexec_fn=preexec_fn,
env=self.env,
close_fds=True,
)
self.started = time.time()
开发者ID:msabramo，项目名称:circus，代码行数:34，代码来源:fly.py
示例15: spawn
def spawn(self):
sockets_fds = self._get_sockets_fds()
args = self.format_args(sockets_fds=sockets_fds)
def preexec_fn():
streams = [sys.stdin]
if self.close_child_stdout:
streams.append(sys.stdout)
if self.close_child_stderr:
streams.append(sys.stderr)
self._null_streams(streams)
os.setsid()
for limit, value in self.rlimits.items():
res = getattr(resource, 'RLIMIT_%s' % limit.upper(), None)
if res is None:
raise ValueError('unknown rlimit "%s"' % limit)
# TODO(petef): support hard/soft limits
resource.setrlimit(res, (value, value))
if self.gid:
try:
os.setgid(self.gid)
except OverflowError:
if not ctypes:
raise
# versions of python < 2.6.2 don't manage unsigned int for
# groups like on osx or fedora
os.setgid(-ctypes.c_int(-self.gid).value)
if self.username is not None:
try:
os.initgroups(self.username, self.gid)
except (OSError, AttributeError):
# not support on Mac or 2.6
pass
if self.uid:
os.setuid(self.uid)
extra = {}
if self.pipe_stdout:
extra['stdout'] = PIPE
if self.pipe_stderr:
extra['stderr'] = PIPE
self._worker = Popen(args, cwd=self.working_dir,
shell=self.shell, preexec_fn=preexec_fn,
env=self.env, close_fds=not self.use_fds,
executable=self.executable, **extra)
# let go of sockets created only for self._worker to inherit
self._sockets = []
self.started = time.time()
开发者ID:IsCoolEntertainment，项目名称:debpkg_circus，代码行数:60，代码来源:process.py
示例16: spawn
def spawn(self):
args = self.format_args()
def preexec_fn():
os.setsid()
for limit, value in self.rlimits.items():
res = getattr(resource, 'RLIMIT_%s' % limit.upper(), None)
if res is None:
raise ValueError('unknown rlimit "%s"' % limit)
# TODO(petef): support hard/soft limits
resource.setrlimit(res, (value, value))
if self.gid:
try:
os.setgid(self.gid)
except OverflowError:
if not ctypes:
raise
# versions of python < 2.6.2 don't manage unsigned int for
# groups like on osx or fedora
os.setgid(-ctypes.c_int(-self.gid).value)
if self.uid:
os.setuid(self.uid)
self._worker = Popen(args, cwd=self.working_dir,
shell=self.shell, preexec_fn=preexec_fn,
env=self.env, close_fds=not self.use_fds,
stdout=PIPE, stderr=PIPE,
executable=self.executable)
self.started = time.time()
开发者ID:andreypopp，项目名称:circus，代码行数:33，代码来源:process.py
示例17: find_device
def find_device(self):
for drive in disk_partitions(False):
if 'removable' in drive[3]:
if os.path.isdir(os.path.join(drive[1], 'system')) and \
os.path.isdir(os.path.join(drive[1], 'documents')):
return drive[1]
# print(self.config['GENERAL']['SSHEnabled'])
if self.config['GENERAL']['SSHEnabled'] == "True":
ssh = Popen('"' + self.config['SSH']['PLinkPath'] + '" [email protected]' + self.config['SSH']['KindleIP']
+ ' whoami', stdout=PIPE, stderr=STDOUT, shell=True)
ssh_check = ssh.wait()
if ssh_check == 0:
self.ssh = True
return self.config['SSH']['KindleIP']
else:
raise OSError('Can\'t connect to Kindle!')
else:
raise OSError('Not found any connected Kindle!')
开发者ID:knigophil，项目名称:KindleWisper，代码行数:19，代码来源:Interface.py
示例18: run
def run(base_dir, timeout):
"""run the program.
return (ok?, msg)"""
cmd = ['./a.out']
out_path = os.path.join(base_dir, 'test.out')
in_path = os.path.join(base_dir, 'test.in')
with open(out_path, 'w') as fout, open(in_path) as fin:
p = Popen(cmd, stdin=fin, stdout=fout, cwd=base_dir)
try:
p.wait(timeout)
except TimeoutExpired:
p.kill()
return False, 'time limit exceed'
else:
if p.returncode == 0:
return True, ''
else:
return False, 'runtime error'
开发者ID:jhonprice，项目名称:yaoj-django，代码行数:19，代码来源:tasks.py
示例19: __init__
def __init__(self, wid, cmd, working_dir=None, shell=False, uid=None,
gid=None, env=None, rlimits=None):
self.wid = wid
if working_dir is None:
self.working_dir = get_working_dir()
else:
self.working_dir = working_dir
self.shell = shell
self.env = env
if rlimits is not None:
self.rlimits = rlimits
else:
self.rlimits = {}
self.cmd = cmd.replace('$WID', str(self.wid))
if uid is None:
self.uid = None
else:
self.uid = to_uid(uid)
if gid is None:
self.gid = None
else:
self.gid = to_gid(gid)
def preexec_fn():
os.setsid()
for limit, value in self.rlimits.items():
res = getattr(resource, 'RLIMIT_%s' % limit.upper(), None)
if res is None:
raise ValueError('unknown rlimit "%s"' % limit)
# TODO(petef): support hard/soft limits
resource.setrlimit(res, (value, value))
if self.gid:
try:
os.setgid(self.gid)
except OverflowError:
if not ctypes:
raise
# versions of python < 2.6.2 don't manage unsigned int for
# groups like on osx or fedora
os.setgid(-ctypes.c_int(-self.gid).value)
if self.uid:
os.setuid(self.uid)
self._worker = Popen(self.cmd.split(), cwd=self.working_dir,
shell=self.shell, preexec_fn=preexec_fn,
env=self.env, close_fds=True, stdout=PIPE,
stderr=PIPE)
self.started = time.time()
开发者ID:cyberj，项目名称:circus，代码行数:54，代码来源:process.py
示例20: _start
def _start(self, ini, xml):
"""start the process"""
if self.isRunning:
return
# write the Halcyon.ini config file
f = open(os.path.join(self.startDir, 'Halcyon.ini'), 'w')
f.write(ini)
f.close()
# write the Regions.cfg file
if not os.path.exists(os.path.join(self.startDir, 'Regions')):
os.mkdir(os.path.join(self.startDir, 'Regions'))
f = open(os.path.join(self.startDir, 'Regions', 'default.xml'), 'w')
f.write(xml)
f.close()
self.proc = Popen(self.startString.split(" "), cwd=self.startDir, stdout=DEVNULL, stderr=DEVNULL)
#write a pidfile
f = open(self.pidFile, 'w')
f.write(str(self.proc.pid))
f.close()
开发者ID:M-O-S-E-S，项目名称:mgm-node，代码行数:22，代码来源:Region.py
注：本文中的psutil.Popen类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论