本文整理汇总了Python中qiime.test.initiate_timeout函数的典型用法代码示例。如果您正苦于以下问题：Python initiate_timeout函数的具体用法？Python initiate_timeout怎么用？Python initiate_timeout使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了initiate_timeout函数的15个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: setUp
def setUp(self):
""" """
self.test_data = get_test_data_fps()
self.files_to_remove = []
self.dirs_to_remove = []
# Create example output directory
tmp_dir = get_qiime_temp_dir()
self.test_out = mkdtemp(dir=tmp_dir,
prefix='core_qiime_analyses_test_',
suffix='')
self.dirs_to_remove.append(self.test_out)
self.qiime_config = load_qiime_config()
self.params = parse_qiime_parameters(params_f1)
# suppress stderr during tests (one of the systems calls in the
# workflow prints a warning, and we can't suppress that warning with
# warnings.filterwarnings) here because it comes from within the code
# executed through the system call. Found this trick here:
# http://stackoverflow.com/questions/9949633/suppressing-print-as-stdout-python
self.saved_stderr = sys.stderr
sys.stderr = StringIO()
initiate_timeout(180)
开发者ID:Kleptobismol，项目名称:qiime，代码行数:25，代码来源:test_downstream.py
示例2: setUp
def setUp(self):
""" """
self.files_to_remove = []
self.dirs_to_remove = []
tmp_dir = get_qiime_temp_dir()
self.test_out = get_tmp_filename(tmp_dir=tmp_dir,
prefix='qiime_parallel_taxonomy_assigner_tests_',
suffix='',
result_constructor=str)
self.dirs_to_remove.append(self.test_out)
create_dir(self.test_out)
self.tmp_seq_filepath = get_tmp_filename(tmp_dir=self.test_out,
prefix='qiime_parallel_taxonomy_assigner_tests_input',
suffix='.fasta')
seq_file = open(self.tmp_seq_filepath, 'w')
seq_file.write(blast_test_seqs.toFasta())
seq_file.close()
self.files_to_remove.append(self.tmp_seq_filepath)
self.id_to_taxonomy_file = NamedTemporaryFile(
prefix='qiime_parallel_taxonomy_assigner_tests_id_to_taxonomy',
suffix='.txt',dir=tmp_dir)
self.id_to_taxonomy_file.write(blast_id_to_taxonomy)
self.id_to_taxonomy_file.seek(0)
self.reference_seqs_file = NamedTemporaryFile(
prefix='qiime_parallel_taxonomy_assigner_tests_ref_seqs',
suffix='.fasta',dir=tmp_dir)
self.reference_seqs_file.write(blast_reference_seqs.toFasta())
self.reference_seqs_file.seek(0)
initiate_timeout(60)
开发者ID:DDomogala3，项目名称:qiime，代码行数:34，代码来源:test_assign_taxonomy.py
示例3: setUp
def setUp(self):
""" """
self.files_to_remove = []
self.dirs_to_remove = []
tmp_dir = get_qiime_temp_dir()
self.test_out = mkdtemp(dir=tmp_dir,
prefix='qiime_parallel_tests_',
suffix='')
self.dirs_to_remove.append(self.test_out)
fd, self.template_fp = mkstemp(dir=self.test_out,
prefix='qiime_template',
suffix='.fasta')
close(fd)
template_f = open(self.template_fp, 'w')
template_f.write(pynast_test1_template_fasta)
template_f.close()
self.files_to_remove.append(self.template_fp)
fd, self.inseqs1_fp = mkstemp(dir=self.test_out,
prefix='qiime_inseqs',
suffix='.fasta')
close(fd)
inseqs1_f = open(self.inseqs1_fp, 'w')
inseqs1_f.write(inseqs1)
inseqs1_f.close()
self.files_to_remove.append(self.inseqs1_fp)
initiate_timeout(60)
开发者ID:Springbudder，项目名称:qiime，代码行数:30，代码来源:test_align_seqs.py
示例4: setUp
def setUp(self):
self.files_to_remove = []
self.dirs_to_remove = []
# Create example output directory
tmp_dir = get_qiime_temp_dir()
self.test_out = get_tmp_filename(tmp_dir=tmp_dir,
prefix='qiime_parallel_tests_',
suffix='',
result_constructor=str)
self.dirs_to_remove.append(self.test_out)
create_dir(self.test_out)
# Create example input file
self.inseqs1_fp = get_tmp_filename(tmp_dir=self.test_out,
prefix='qiime_inseqs',
suffix='.fasta')
inseqs1_f = open(self.inseqs1_fp,'w')
inseqs1_f.write(inseqs1)
inseqs1_f.close()
self.files_to_remove.append(self.inseqs1_fp)
# Define number of seconds a test can run for before timing out
# and failing
initiate_timeout(60)
开发者ID:B-Rich，项目名称:cmd-abstraction，代码行数:26，代码来源:test_util.py
示例5: setUp
def setUp(self):
self.files_to_remove = []
self.dirs_to_remove = []
# Create example output directory
tmp_dir = get_qiime_temp_dir()
self.test_out = mkdtemp(dir=tmp_dir,
prefix='core_qiime_analyses_test_',
suffix='')
self.dirs_to_remove.append(self.test_out)
# Get input data
self.test_data = get_test_data_fps()
self.qiime_config = load_qiime_config()
self.qiime_config['jobs_to_start'] = 2
self.qiime_config['seconds_to_sleep'] = 1
# suppress stderr during tests (one of the systems calls in the
# workflow prints a warning, and we can't suppress that warning with
# warnings.filterwarnings) here because it comes from within the code
# executed through the system call. Found this trick here:
# http://stackoverflow.com/questions/9949633/suppressing-print-as-stdout-python
self.saved_stderr = sys.stderr
sys.stderr = StringIO()
# Define number of seconds a test can run for before timing out
# and failing
initiate_timeout(600)
开发者ID:Springbudder，项目名称:qiime，代码行数:30，代码来源:test_core_diversity_analyses.py
示例6: setUp
def setUp(self):
""" """
self.files_to_remove = []
self.dirs_to_remove = []
tmp_dir = get_qiime_temp_dir()
self.test_out = mkdtemp(dir=tmp_dir,
prefix='qiime_parallel_blaster_tests_',
suffix='')
self.dirs_to_remove.append(self.test_out)
fd, self.tmp_seq_filepath = mkstemp(dir=self.test_out,
prefix='qiime_parallel_blaster_tests_input',
suffix='.fasta')
close(fd)
seq_file = open(self.tmp_seq_filepath, 'w')
seq_file.write(blast_test_seqs)
seq_file.close()
self.files_to_remove.append(self.tmp_seq_filepath)
self.reference_seqs_file = NamedTemporaryFile(
prefix='qiime_parallel_blaster_tests_ref_seqs',
suffix='.fasta', dir=tmp_dir)
self.reference_seqs_file.write(blast_ref_seqs)
self.reference_seqs_file.seek(0)
initiate_timeout(60)
开发者ID:Kleptobismol，项目名称:qiime，代码行数:27，代码来源:test_blast.py
示例7: setUp
def setUp(self):
""" """
self.files_to_remove = []
self.dirs_to_remove = []
tmp_dir = get_qiime_temp_dir()
self.test_out = get_tmp_filename(tmp_dir=tmp_dir,
prefix='qiime_parallel_tests_',
suffix='',
result_constructor=str)
self.dirs_to_remove.append(self.test_out)
create_dir(self.test_out)
self.template_fp = get_tmp_filename(tmp_dir=self.test_out,
prefix='qiime_template',
suffix='.fasta')
template_f = open(self.template_fp,'w')
template_f.write(pynast_test1_template_fasta)
template_f.close()
self.files_to_remove.append(self.template_fp)
self.inseqs1_fp = get_tmp_filename(tmp_dir=self.test_out,
prefix='qiime_inseqs',
suffix='.fasta')
inseqs1_f = open(self.inseqs1_fp,'w')
inseqs1_f.write(inseqs1)
inseqs1_f.close()
self.files_to_remove.append(self.inseqs1_fp)
initiate_timeout(60)
开发者ID:qiime，项目名称:qp-refactor，代码行数:30，代码来源:test_align_seqs.py
示例8: setUp
def setUp(self):
"""
"""
tmp_dir = get_qiime_temp_dir()
self.test_out = mkdtemp(dir=tmp_dir,
prefix='qiime_parallel_tests_',
suffix='')
self.dirs_to_remove = [self.test_out]
self.output_fp = join(self.test_out, 'fmap.txt')
self.failure_fp = join(self.test_out, 'fail.txt')
self.usearch_fp = join(self.test_out, 'out.uc')
self.bl6_fp = join(self.test_out, 'out.bl6')
self.log_fp = join(self.test_out, 'fmap.log')
self.files_to_remove = [self.output_fp, self.failure_fp,
self.usearch_fp, self.log_fp, self.bl6_fp]
fd, self.refseqs1_fp = mkstemp(dir=self.test_out,
prefix='qiime_refseqs',
suffix='.fasta')
close(fd)
refseqs1_f = open(self.refseqs1_fp, 'w')
refseqs1_f.write(refseqs1)
refseqs1_f.close()
self.files_to_remove.append(self.refseqs1_fp)
fd, self.refseqs2_fp = mkstemp(dir=self.test_out,
prefix='qiime_refseqs',
suffix='.fasta')
close(fd)
refseqs2_f = open(self.refseqs2_fp, 'w')
refseqs2_f.write(refseqs2)
refseqs2_f.close()
self.files_to_remove.append(self.refseqs2_fp)
fd, self.inseqs1_fp = mkstemp(dir=self.test_out,
prefix='qiime_inseqs',
suffix='.fasta')
close(fd)
inseqs1_f = open(self.inseqs1_fp, 'w')
inseqs1_f.write(inseqs1)
inseqs1_f.close()
self.files_to_remove.append(self.inseqs1_fp)
fd, self.inseqs2_fp = mkstemp(dir=self.test_out,
prefix='qiime_inseqs',
suffix='.fasta')
close(fd)
inseqs2_f = open(self.inseqs2_fp, 'w')
inseqs2_f.write(inseqs2)
inseqs2_f.close()
self.files_to_remove.append(self.inseqs2_fp)
initiate_timeout(60)
开发者ID:ElDeveloper，项目名称:qiime，代码行数:52，代码来源:test_map_reads_to_reference.py
示例9: setUp
def setUp(self):
"""
"""
tmp_dir = get_qiime_temp_dir()
self.test_out = get_tmp_filename(tmp_dir=tmp_dir,
prefix='qiime_parallel_tests_',
suffix='',
result_constructor=str)
create_dir(self.test_out)
self.dirs_to_remove = [self.test_out]
self.output_fp = join(self.test_out,'fmap.txt')
self.failure_fp = join(self.test_out,'fail.txt')
self.usearch_fp = join(self.test_out,'out.uc')
self.bl6_fp = join(self.test_out,'out.bl6')
self.log_fp = join(self.test_out,'fmap.log')
self.files_to_remove = [self.output_fp, self.failure_fp,
self.usearch_fp, self.log_fp, self.bl6_fp]
self.refseqs1_fp = get_tmp_filename(tmp_dir=self.test_out,
prefix='qiime_refseqs',
suffix='.fasta')
refseqs1_f = open(self.refseqs1_fp,'w')
refseqs1_f.write(refseqs1)
refseqs1_f.close()
self.files_to_remove.append(self.refseqs1_fp)
self.refseqs2_fp = get_tmp_filename(tmp_dir=self.test_out,
prefix='qiime_refseqs',
suffix='.fasta')
refseqs2_f = open(self.refseqs2_fp,'w')
refseqs2_f.write(refseqs2)
refseqs2_f.close()
self.files_to_remove.append(self.refseqs2_fp)
self.inseqs1_fp = get_tmp_filename(tmp_dir=self.test_out,
prefix='qiime_inseqs',
suffix='.fasta')
inseqs1_f = open(self.inseqs1_fp,'w')
inseqs1_f.write(inseqs1)
inseqs1_f.close()
self.files_to_remove.append(self.inseqs1_fp)
self.inseqs2_fp = get_tmp_filename(tmp_dir=self.test_out,
prefix='qiime_inseqs',
suffix='.fasta')
inseqs2_f = open(self.inseqs2_fp,'w')
inseqs2_f.write(inseqs2)
inseqs2_f.close()
self.files_to_remove.append(self.inseqs2_fp)
initiate_timeout(60)
开发者ID:DDomogala3，项目名称:qiime，代码行数:50，代码来源:test_map_reads_to_reference.py
示例10: setUp
def setUp(self):
"""Set up files/environment that will be used by the tests."""
# The prefix to use for temporary files. This prefix may be added to,
# but all temp dirs and files created by the tests will have this
# prefix at a minimum.
self.prefix = 'generate_taxa_compare_table_tests'
self.start_dir = getcwd()
self.dirs_to_remove = []
self.files_to_remove = []
self.tmp_dir = get_qiime_temp_dir()
if not exists(self.tmp_dir):
makedirs(self.tmp_dir)
# if test creates the temp dir, also remove it
self.dirs_to_remove.append(self.tmp_dir)
# setup temporary root input directory
self.root_dir = mkdtemp(dir=self.tmp_dir,
prefix='%s_root_dir_' %self.prefix)
self.dirs_to_remove.append(self.root_dir)
L18S_dir = '/L18S-1/blast_1.0/'
makedirs(self.root_dir+L18S_dir)
self.L18S_fp = self.root_dir+L18S_dir+'/otu_table_mc2_no_pynast_failures_w_taxa_L5.txt'
with open(self.L18S_fp, 'w') as f:
f.writelines(L18S_L5_blast_one_multiple_assign_output)
self.files_to_remove.append(self.L18S_fp)
# setup temporary key directory
self.key_dir = mkdtemp(dir=self.tmp_dir,
prefix='%s_key_dir_' %self.prefix)
self.dirs_to_remove.append(self.key_dir)
self.key_fp = self.key_dir+'/L18S_key.txt'
with open(self.key_fp, 'w') as f:
f.writelines(L18S_key)
self.files_to_remove.append(self.key_fp)
self.bad_key = self.key_dir+'/L18S_key.txt'
# setup temporary output directory
self.output_dir = mkdtemp(dir=self.tmp_dir,
prefix='%s_output_dir_' %self.prefix)
self.dirs_to_remove.append(self.output_dir)
self.ts1 = parse_taxa_summary_table(ts1.split('\n'))
initiate_timeout(60)
开发者ID:ebolyen，项目名称:short-read-tax-assignment，代码行数:47，代码来源:test_generate_taxa_compare_table.py
示例11: setUp
def setUp(self):
""" """
self.test_data = get_test_data_fps()
self.files_to_remove = []
self.dirs_to_remove = []
# Create example output directory
tmp_dir = get_qiime_temp_dir()
self.test_out = mkdtemp(dir=tmp_dir,
prefix='core_qiime_analyses_test_',
suffix='')
self.dirs_to_remove.append(self.test_out)
self.qiime_config = load_qiime_config()
self.params = parse_qiime_parameters([])
initiate_timeout(60)
开发者ID:Bonder-MJ，项目名称:qiime，代码行数:17，代码来源:test_upstream.py
示例12: setUp
def setUp(self):
"""Set up files/environment that will be used by the tests."""
# The prefix to use for temporary files. This prefix may be added to,
# but all temp dirs and files created by the tests will have this
# prefix at a minimum.
self.prefix = 'tax2tree_controller_tests'
self.start_dir = getcwd()
self.dirs_to_remove = []
self.files_to_remove = []
self.tmp_dir = get_qiime_temp_dir()
if not exists(self.tmp_dir):
makedirs(self.tmp_dir)
# if test creates the temp dir, also remove it
self.dirs_to_remove.append(self.tmp_dir)
initiate_timeout(60)
开发者ID:rob-knight，项目名称:qiime，代码行数:18，代码来源:test_tax2tree.py
示例13: setUp
def setUp(self):
""" """
self.test_data = get_test_data_fps()
self.files_to_remove = []
self.dirs_to_remove = []
# Create example output directory
tmp_dir = get_qiime_temp_dir()
self.test_out = mkdtemp(dir=tmp_dir,
prefix='core_qiime_analyses_test_',
suffix='')
self.dirs_to_remove.append(self.test_out)
self.qiime_config = load_qiime_config()
self.params = parse_qiime_parameters([])
self.params_sortmerna = parse_qiime_parameters(
['pick_otus:otu_picking_method\tsortmerna'])
self.params_sumaclust = parse_qiime_parameters(
['pick_otus:otu_picking_method\tsumaclust'])
initiate_timeout(60)
开发者ID:nbresnick，项目名称:qiime，代码行数:21，代码来源:test_upstream.py
示例14: setUp
def setUp(self):
""" """
self.files_to_remove = []
self.dirs_to_remove = []
tmp_dir = get_qiime_temp_dir()
self.test_out = mkdtemp(dir=tmp_dir, prefix="qiime_parallel_tests_", suffix="")
self.dirs_to_remove.append(self.test_out)
fd, self.refseqs1_fp = mkstemp(dir=self.test_out, prefix="qiime_refseqs", suffix=".fasta")
close(fd)
refseqs1_f = open(self.refseqs1_fp, "w")
refseqs1_f.write(refseqs1)
refseqs1_f.close()
self.files_to_remove.append(self.refseqs1_fp)
fd, self.refseqs2_fp = mkstemp(dir=self.test_out, prefix="qiime_refseqs", suffix=".fasta")
close(fd)
refseqs2_f = open(self.refseqs2_fp, "w")
refseqs2_f.write(refseqs2)
refseqs2_f.close()
self.files_to_remove.append(self.refseqs2_fp)
fd, self.inseqs1_fp = mkstemp(dir=self.test_out, prefix="qiime_inseqs", suffix=".fasta")
close(fd)
inseqs1_f = open(self.inseqs1_fp, "w")
inseqs1_f.write(inseqs1)
inseqs1_f.close()
self.files_to_remove.append(self.inseqs1_fp)
fd, self.inseqs2_fp = mkstemp(dir=self.test_out, prefix="qiime_inseqs", suffix=".fasta")
close(fd)
inseqs2_f = open(self.inseqs2_fp, "w")
inseqs2_f.write(inseqs2)
inseqs2_f.close()
self.files_to_remove.append(self.inseqs2_fp)
initiate_timeout(60)
开发者ID:Honglongwu，项目名称:qiime，代码行数:38，代码来源:test_map_reads_to_reference.py
示例15: setUp
def setUp(self):
"""Set up files/environment that will be used by the tests."""
# The prefix to use for temporary files. This prefix may be added to,
# but all temp dirs and files created by the tests will have this
# prefix at a minimum.
self.prefix = 'multiple_assign_taxonomy_tests'
self.start_dir = getcwd()
self.dirs_to_remove = []
self.files_to_remove = []
self.tmp_dir = get_qiime_temp_dir()
if not exists(self.tmp_dir):
makedirs(self.tmp_dir)
# if test creates the temp dir, also remove it
self.dirs_to_remove.append(self.tmp_dir)
# setup temporary output directories
self.output_dir = mkdtemp(dir=self.tmp_dir,
prefix='%s_output_dir_' %self.prefix)
self.dirs_to_remove.append(self.output_dir)
initiate_timeout(60)
开发者ID:zellett，项目名称:short-read-tax-assignment-mirror，代码行数:23，代码来源:test_multiple_assign_taxonomy.py
注：本文中的qiime.test.initiate_timeout函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论