本文整理汇总了Python中subprocess32.check_output函数的典型用法代码示例。如果您正苦于以下问题:Python check_output函数的具体用法?Python check_output怎么用?Python check_output使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了check_output函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: __init__
def __init__(self,platform_directory_string=None,root_directory_string=None,ssh_alias="",remote=False,hostname=None,shell_vars={},shell_setup_cmds=[],shell_exit_cmds=[]):
"""Constructor
Parameters
platform_directory_string - (string) location of platform specific code
root_directory_string - (string) location of F^3 on this system
ssh_alias - (string) SSH alias for this system, as stored in .ssh/sshconfig
remote - (bool) Is this system remote?
hostname - (string) - override the system hostname for generated code
shell_vars - (dict) - set any environmental variables
shell_setup_cmds (list of strings) - any commands that need to be run upon login
shell_exit_cmds (list of strings) - any commands that need to be run upon logout
Upon construction, if this is a remote system, all of the SSH set up is done. It might take several seconds to return.
"""
self.platform_directory_string = platform_directory_string
self.root_directory_string = root_directory_string
self.ssh_alias = ssh_alias
self.remote = remote
self.hostname = hostname
self.shell_vars = {}
for k in shell_vars: self.shell_vars[k] = shell_vars[k]
self.shell_setup_cmds = shell_setup_cmds
self.shell_exit_cmds = shell_exit_cmds
if(self.remote and not(self.ssh_alias)): raise ValueError("ssh alias needs to be set")
#in case the environmental variable isn't set
if not(self.root_directory_string) and not(self.remote):
try: self.root_directory_string = os.environ["F3_ROOT"]
except KeyError:
print("F3_ROOT environmental variable not set")
sys.exit(1)
#if the environmental variable isn't set, and its on a remote server
elif not(self.root_directory_string and self.remote):
try:
ssh_cmd = ["ssh","%s"%self.ssh_alias,"source",".profile;","printenv","|","grep","^F3_ROOT"]
output = subprocess.check_output(ssh_cmd)
if not(output): raise KeyError("F3_ROOT environmental variable not set on %s"%self.ssh_alias)
output= output.split("=")[1].strip("\r\n")
self.root_directory_string = output
except KeyError:
print("F3_ROOT environmental variable not set on %s"%self.ssh_alias)
sys.exit(1)
if(self.hostname==None and not(self.remote)):
self.hostname = os.uname()[1].replace(".","_") #System hostname
elif(self.hostname==None):
ssh_cmd = ["ssh","%s"%self.ssh_alias,"hostname"]
output = subprocess.check_output(ssh_cmd)
#if not(output): raise KeyError("F3_ROOT environmental variable not set on %s"%self.ssh_alias)
output = output.replace("\n","") #output.split("=")[1].strip("\r\n")
self.hostname = output.replace(".","_")
开发者ID:Gordonei,项目名称:ForwardFinancialFramework,代码行数:60,代码来源:Platform.py
示例2: hashes
def hashes(repo_dir):
hashes_ = dict(master='', feature='', tag_annotated='', tag_light='')
hashes_['master'] = subprocess32.check_output(['git', 'rev-parse', 'HEAD'], cwd=repo_dir).strip().decode('ascii')
subprocess32.check_call(['git', 'checkout', '-b', 'feature'], cwd=repo_dir)
with open(os.path.join(repo_dir, 'test.txt'), 'a') as f:
f.write('test')
subprocess32.check_call(['git', 'add', 'test.txt'], cwd=repo_dir)
subprocess32.check_call(['git', 'commit', '-m', 'Wrote to file.'], cwd=repo_dir)
hashes_['feature'] = subprocess32.check_output(['git', 'rev-parse', 'HEAD'], cwd=repo_dir).strip().decode('ascii')
subprocess32.check_call(['git', 'checkout', 'master'], cwd=repo_dir)
with open(os.path.join(repo_dir, 'test.txt'), 'a') as f:
f.write('test2')
subprocess32.check_call(['git', 'add', 'test.txt'], cwd=repo_dir)
subprocess32.check_call(['git', 'commit', '-m', 'Wrote to file2.'], cwd=repo_dir)
subprocess32.check_call(['git', 'tag', '-a', 'v1.0', '-m', 'First Version'], cwd=repo_dir)
hashes_['tag_annotated'] = subprocess32.check_output(['git', 'rev-parse', 'HEAD'],
cwd=repo_dir).strip().decode('ascii')
with open(os.path.join(repo_dir, 'test.txt'), 'a') as f:
f.write('test3')
subprocess32.check_call(['git', 'add', 'test.txt'], cwd=repo_dir)
subprocess32.check_call(['git', 'commit', '-m', 'Wrote to file3.'], cwd=repo_dir)
subprocess32.check_call(['git', 'tag', 'v1.0l'], cwd=repo_dir)
hashes_['tag_light'] = subprocess32.check_output(['git', 'rev-parse', 'HEAD'], cwd=repo_dir).strip().decode('ascii')
assert all(hashes_.values())
assert 4 == len(set(hashes_.values()))
return hashes_
开发者ID:Robpol86,项目名称:coveralls-multi-ci,代码行数:30,代码来源:conftest.py
示例3: generate
def generate(self,name_extension=".c",override=True,verbose=False,debug=False):
#os.chdir("..")
#os.chdir(self.platform.platform_directory())
if(override or not os.path.exists("%s/%s%s"%(os.path.join(self.platform.root_directory(),self.platform.platform_directory()),self.output_file_name,name_extension))):
#os.chdir(self.platform.root_directory())
#os.chdir("bin")
code_string = []
code_string.extend(self.generate_identifier())
code_string.extend(self.generate_libraries())
code_string.extend(self.generate_variable_declaration())
code_string.extend(self.generate_activity_thread())
code_string.extend(self.generate_main_thread())
#Actually writing to the file
self.generate_source(code_string,name_extension,verbose,debug)
#If remote connection
if(self.platform.remote):
copy_command = ["scp"]
copy_command += ["%s/%s%s"%(os.path.join(self.platform.root_directory(),self.platform.platform_directory()),self.output_file_name,name_extension)]
copy_command += ["%s:/home/gordon/workspace/ForwardFinancialFramework/%s"%(self.platform.ssh_alias,self.platform.platform_directory())]
subprocess.check_output(copy_command)
if(debug): print("Copied %s to %s"%(copy_command[1],copy_command[-1]))
开发者ID:biddyweb,项目名称:ForwardFinancialFramework,代码行数:25,代码来源:MulticoreCPU_MonteCarlo.py
示例4: attach
def attach(self):
if self.attached:
raise Exception("Wrong usage. Ramdisk should not be attached")
if not check_is_same_device_as_root_fs(self.folder):
msg = ('Folder must be on / '
'(ROOT) and must not be a different device (possibly '
'already a RAMDISK)! Maybe try "umount {0}"?').format(self.folder)
raise argparse.ArgumentTypeError(msg)
create_ramdisk_stdout = subprocess.check_output(
['hdid','-nomount', 'ram://{0}'.format(self.size)])
ram_disk_device = create_ramdisk_stdout.strip().strip('\n')
check_is_not_normal_harddrive(ram_disk_device)
logger.info('Created RAM disk {0}'.format(ram_disk_device))
logger.info('Formatting RAM disk...')
format_stdout = subprocess.check_output(['newfs_hfs', ram_disk_device])
#Initialized /dev/rdisk13 as a 512 MB HFS Plus volume
assert format_stdout, format_stdout
old_ionode_nbr = os.stat(self.folder).st_ino
logger.info('Mounting RAM disk {0} as {1}'.format(ram_disk_device, self.folder))
subprocess.check_call(['mount','-t','hfs', ram_disk_device, self.folder])
assert old_ionode_nbr != os.stat(self.folder).st_ino
# TODO: probably remove this
assert not check_is_same_device_as_root_fs(self.folder)
self.ram_disk_device = ram_disk_device
self.attached = True
开发者ID:binarytemple,项目名称:ramdisk-mounter,代码行数:33,代码来源:ramdisk_mounter.py
示例5: _delete_docker_interface
def _delete_docker_interface(self):
"""Delete the existing veth connecting to the docker bridge."""
logger.debug('Deleting docker interface eth0')
# Get the PID of the container.
pid = str(self._get_container_pid(self.docker_id))
logger.debug('Container %s running with PID %s', self.docker_id, pid)
# Set up a link to the container's netns.
logger.debug("Linking to container's netns")
logger.debug(check_output(['mkdir', '-p', '/var/run/netns']))
netns_file = '/var/run/netns/' + pid
if not os.path.isfile(netns_file):
logger.debug(check_output(['ln', '-s', '/proc/' + pid + '/ns/net',
netns_file]))
# Log our container's interfaces before making any changes.
_log_interfaces(pid)
# Reach into the netns and delete the docker-allocated interface.
logger.debug(check_output(['ip', 'netns', 'exec', pid,
'ip', 'link', 'del', 'eth0']))
# Log our container's interfaces after making our changes.
_log_interfaces(pid)
# Clean up after ourselves (don't want to leak netns files)
logger.debug(check_output(['rm', netns_file]))
开发者ID:fasaxc,项目名称:calico-kubernetes,代码行数:28,代码来源:calico_kubernetes.py
示例6: get_from_requirements
def get_from_requirements(self,
r, # type: Optional[Dict[Text, Text]]
req, # type: bool
pull_image, # type: bool
force_pull=False, # type: bool
tmp_outdir_prefix=None # type: Text
):
# type: (...) -> Optional[Text]
"""
Returns the filename of the Singularity image (e.g.
hello-world-latest.img).
"""
if r:
errmsg = None
try:
check_output(["singularity", "--version"])
except CalledProcessError as err:
errmsg = "Cannot execute 'singularity --version' {}".format(err)
except OSError as err:
errmsg = "'singularity' executable not found: {}".format(err)
if errmsg:
if req:
raise WorkflowException(errmsg)
else:
return None
if self.get_image(r, pull_image, force_pull):
return os.path.abspath(r["dockerImageId"])
if req:
raise WorkflowException(u"Container image {} not "
"found".format(r["dockerImageId"]))
return None
开发者ID:pvanheus,项目名称:cwltool,代码行数:35,代码来源:singularity.py
示例7: run_two_test
def run_two_test(test):
binname = ['./{0}'.format(test), 'input.txt', 'in.txt']
try:
output = subprocess.check_output(binname, stderr=FNULL, timeout=30.0)
except subprocess.CalledProcessError as e:
output = "Program Error: {0}\n score is 0.0\n".format(e)
if int(format(e).split(' ')[-1])<0:
print output
return output, get_score(output)
except subprocess.TimeoutExpired:
output = "Execution Timed out: score is 0.0\n"
return output, get_score(output)
except OSError:
output = "Compilation failed: score is 0.0\n"
return output, get_score(output)
binname = ['python', '{0}.py'.format(test), '-i', 'out_{0}.txt'.format(test), '-o', 'in.txt'.format(test)]
try:
output = subprocess.check_output(binname, timeout=30.0)
except subprocess.CalledProcessError:
output = "Program Error (Output file not in expected format): score is 0.0\n"
except subprocess.TimeoutExpired:
output = "Execution Timed out: score is 0.0\n"
except OSError:
output = "Compilation failed: score is 0.0\n"
print output
return output, get_score(output)
开发者ID:akumark,项目名称:hocus-pocus,代码行数:26,代码来源:main_backup.py
示例8: generate
def generate(self,name_extension=".c",override=True,verbose=False,debug=False):
"""Code generation method
Parameters
name_extension - (string) file name extension to use for generated code
override - (bool) option to force code generation
verbose - (bool) option for setting verbosity level of code generation
debug - (bool) option passed to source code generation method
"""
if(override or not os.path.exists("%s/%s%s"%(os.path.join(self.platform.root_directory(),self.platform.platform_directory()),self.output_file_name,name_extension))):
code_string = []
code_string.extend(self.generate_identifier())
code_string.extend(self.generate_libraries())
code_string.extend(self.generate_variable_declaration())
code_string.extend(self.generate_activity_thread(debug))
code_string.extend(self.generate_main_thread())
#Actually writing to the file
self.generate_source(code_string,name_extension,verbose,debug)
#If remote connection
if(self.platform.remote):
copy_command = ["scp"]
copy_command += ["%s/%s%s"%(os.path.join(self.platform.root_directory(),self.platform.platform_directory()),self.output_file_name,name_extension)]
copy_command += ["%s:/home/gordon/workspace/ForwardFinancialFramework/%s"%(self.platform.ssh_alias,self.platform.platform_directory())]
subprocess.check_output(copy_command)
if(debug): print("Copied %s to %s"%(copy_command[1],copy_command[-1]))
开发者ID:Gordonei,项目名称:ForwardFinancialFramework,代码行数:28,代码来源:MulticoreCPU_MonteCarlo.py
示例9: run
def run(self):
self.prepareInputFiles()
bin_path = sample_confs.Path('')
cmds = ['perl', bin_path.ehalf_box_size_bin(), self.lig_sdf]
stdout = subprocess32.check_output(cmds)
box_size, x, y, z = stdout.split()
cmd = '''%s --receptor %s --ligand %s \
--center_x %s\
--center_y %s\
--center_z %s\
--size_x %s\
--size_y %s\
--size_z %s\
--randomize_only\
--cpu 1\
--out %s
''' % (bin_path.vina_bin(), self.prtPdbqt, self.lig_pdbqt, x, y, z,
box_size, box_size, box_size, self.output().path)
print cmd
vina_out = subprocess32.check_output(shlex.split(cmd))
ofn = self.output().path + ".rnd.txt"
with open(ofn, 'w') as ofs:
ofs.write(vina_out)
开发者ID:EricTing,项目名称:extended-contact-mode-score,代码行数:25,代码来源:vina_predict_biolip.py
示例10: run
def run(self):
self.removeSdfs()
cmds = ['bash', self.trace_bin(),
self.sdf_id,
self.requires().output().path]
_ = subprocess32.check_output(cmds)
path_task = Path(self.sdf_id)
pdb_path = path_task.pdb_path()
sdfs = self.getSdfs()
with open(self.output().path, 'w') as ofs:
if len(sdfs) > 0:
count = 0
while count < len(sdfs):
sdf1 = random.choice(sdfs)
sdf2 = random.choice(sdfs)
cmds = ['cms', '-frc',
'--lig1', sdf1,
'--lig2', sdf2,
'--prt1', pdb_path,
'--prt2', pdb_path]
if sdf1 != sdf2:
try:
stdout = subprocess32.check_output(cmds)
ofs.write(stdout)
count += 1
except:
pass
else:
pass
开发者ID:EricTing,项目名称:extended-contact-mode-score,代码行数:31,代码来源:cms_vals_for_p_val.py
示例11: _update_slurm_node
def _update_slurm_node(self, nodename, updates):
cmd = ['scontrol', 'update', 'NodeName=' + nodename] + updates
try:
subprocess.check_output(cmd)
except:
self._logger.error(
"SLURM update %r failed", cmd, exc_info=True)
开发者ID:chapmanb,项目名称:arvados,代码行数:7,代码来源:slurm.py
示例12: test
def test(suffix):
"""Test with subprocess."""
path = os.path.realpath(os.path.join(os.path.dirname(__file__), '..', 'example{0}.py'.format(suffix)))
env = dict(PYTHONIOENCODING='utf-8')
if 'SystemRoot' in os.environ:
env['SystemRoot'] = os.environ['SystemRoot']
subprocess.check_output([sys.executable, path], env=env, stderr=subprocess.STDOUT)
开发者ID:alexfalcucci-archive,项目名称:terminaltables,代码行数:7,代码来源:test_examples.py
示例13: _init_instruments
def _init_instruments(self, bundle_id):
self._bootstrap = os.path.join(__dir__, 'bootstrap.sh')
self._bundle_id = bundle_id
self._env = {'UDID': self.udid, 'BUNDLE_ID': self._bundle_id}
# 1. remove pipe
subprocess.check_output([self._bootstrap, 'reset'], env=self._env)
# 2. start instruments
self._proc = subprocess.Popen([self._bootstrap, 'instruments'], env=self._env, stdout=subprocess.PIPE)
开发者ID:tazjel,项目名称:AutomatorX,代码行数:8,代码来源:__init__.py
示例14: test_app_force_stop
def test_app_force_stop(appiumSetup, moduleSetup, testSetup):
print 'in test app force stop'
subprocess32.call(['adb', '-s', config.udid, 'shell', 'am', 'force-stop', config.app_package])
app_process = subprocess32.check_output(['adb', '-s', config.udid, 'shell', 'ps', '|', 'grep', config.app_package])
assert not app_process, 'app is not forced stopped, process = %s' % app_process
subprocess32.call(['adb', '-s', config.udid, 'shell', 'monkey','-p', config.app_package, '-c', 'android.intent.category.LAUNCHER 1'])
config.appium_driver.background_app(5)
focused_app = subprocess32.check_output(['adb', '-s', config.udid, 'shell', 'dumpsys window windows', '|', 'grep', '-E', 'mCurrentFocus'])
assert config.app_package in focused_app, 'app package %s, not found in %s' % (config.app_package, focused_app)
开发者ID:skapil,项目名称:appmbtest,代码行数:9,代码来源:tests_sanity.py
示例15: set_veth_mac
def set_veth_mac(veth_name_host, mac):
"""
Set the veth MAC address.
:param veth_name_host: The name of the veth.
:param mac: The MAC address.
:return: None. Raises CalledProcessError on error.
"""
# TODO MAC should be an EUI object.
check_output(["ip", "link", "set", "dev", veth_name_host, "address", mac], timeout=IP_CMD_TIMEOUT)
开发者ID:projectcalico,项目名称:libcalico,代码行数:9,代码来源:netns.py
示例16: testProcessName
def testProcessName( self ) :
process = subprocess.Popen( [ "gaffer", "env", "sleep", "100" ] )
time.sleep( 1 )
command = subprocess.check_output( [ "ps", "-p", str( process.pid ), "-o", "command=" ] ).strip()
name = subprocess.check_output( [ "ps", "-p", str( process.pid ), "-o", "comm=" ] ).strip()
process.kill()
self.assertEqual( command, "gaffer env sleep 100" )
self.assertEqual( name, "gaffer" )
开发者ID:lucienfostier,项目名称:gaffer,代码行数:10,代码来源:ApplicationTest.py
示例17: app_version
def app_version(config):
import hashlib
import os
import subprocess
version = subprocess.check_output(["git", "-C", os.path.dirname(__file__), "describe"]).decode("utf-8").strip()
diff = subprocess.check_output(["git", "-C", os.path.dirname(__file__), "diff", "--no-ext-diff"])
if diff:
version += "-patch" + hashlib.sha1(diff).hexdigest()[:7]
config.registry.settings["snovault.app_version"] = version
开发者ID:ENCODE-DCC,项目名称:encoded,代码行数:10,代码来源:__init__.py
示例18: app_version
def app_version(config):
import hashlib
import os
import subprocess
version = subprocess.check_output(
['git', '-C', os.path.dirname(__file__), 'describe']).decode('utf-8').strip()
diff = subprocess.check_output(
['git', '-C', os.path.dirname(__file__), 'diff', '--no-ext-diff'])
if diff:
version += '-patch' + hashlib.sha1(diff).hexdigest()[:7]
config.registry.settings['snovault.app_version'] = version
开发者ID:detrout,项目名称:encoded,代码行数:11,代码来源:__init__.py
示例19: test_bluetooth_on_off
def test_bluetooth_on_off(appiumSetup, moduleSetup, testSetup):
print 'in test bluetooth on off'
tests_app_android.bluetooth_toggle()
tests_app_android.bringAppToForeground()
time.sleep(2)
focused_app = subprocess32.check_output(['adb', '-s', config.udid, 'shell', 'dumpsys window windows', '|', 'grep', '-E', 'mCurrentFocus'])
assert config.app_package in focused_app, 'app package %s, not found in %s' % (config.app_package, focused_app)
tests_app_android.bluetooth_toggle()
tests_app_android.bringAppToForeground()
time.sleep(2)
focused_app = subprocess32.check_output(['adb', '-s', config.udid, 'shell', 'dumpsys window windows', '|', 'grep', '-E', 'mCurrentFocus'])
assert config.app_package in focused_app, 'app package %s, not found in %s' % (config.app_package, focused_app)
开发者ID:skapil,项目名称:appmbtest,代码行数:12,代码来源:tests_sanity.py
示例20: _check_tool
def _check_tool(tool_name):
"""
Checks if the invoked tool produces a known command.
Returns the tool's name if the check is successful,
or a tuple with the tool's name and the error if it's not.
"""
try:
subprocess.check_output([tool_name, '-h'], timeout=TIMEOUT)
return tool_name
except (subprocess.CalledProcessError, OSError) as err:
logger.error(err)
return tool_name, err
开发者ID:adobe-type-tools,项目名称:afdko,代码行数:12,代码来源:runner.py
注:本文中的subprocess32.check_output函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论